Minimum throughput body timeouts Pt.1 (#3068)

## 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 -->
#1562

## Description
<!--- Describe your changes in detail -->
This change adds a new body wrapper: The minimum throughput limit
wrapper. It tracks the rate that data is being streamed from itself. If
that rate falls below some configurable limit, it emits an error instead
of the next chunk. This protects users from requests that start quickly
but then slow down considerably.

I'd like to get this merged and then figure out the
codegen/docs/examples/config part in a separate PR.

## Testing
<!--- Please describe in detail how you tested your changes -->
<!--- Include details of your testing environment, and the tests you ran
to -->
<!--- see how your change affects other areas of the code, etc. -->
Tests are included.

----

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

aws/sdk/integration-tests/s3/Cargo.toml

@@ -36,7 +36,8 @@ futures-util = { version = "0.3.16", default-features = false }
 hdrhistogram = "7.5.2"
 http = "0.2.3"
 http-body = "0.4.5"
-hyper = "0.14.26"
+hyper = { version = "0.14.26", features = ["stream"] }
+once_cell = "1.18.0"
 pretty_assertions = "1.3"
 serde_json = "1"
 smol = "1.2"

rust-runtime/Cargo.toml

@@ -1,6 +1,4 @@
 [workspace]
-
-
 members = [
     "inlineable",
     "aws-smithy-async",

rust-runtime/aws-smithy-async/src/future/pagination_stream.rs

@@ -184,11 +184,11 @@ mod test {
         // `tokio_test::task::Spawn::poll_next` can only be invoked when the wrapped
         // type implements the `Stream` trait. Here, `FnStream` does not implement it,
         // so we work around it by using the `enter` method.
-        let _ = test_stream.enter(|ctx, pin| {
+        test_stream.enter(|ctx, pin| {
             let polled = pin.poll_next(ctx);
             assert!(polled.is_pending());
         });
-        let _ = test_stream.enter(|ctx, pin| {
+        test_stream.enter(|ctx, pin| {
             let polled = pin.poll_next(ctx);
             assert!(polled.is_pending());
         });

rust-runtime/aws-smithy-async/src/rt/sleep.rs

@@ -83,7 +83,7 @@ pub fn default_async_sleep() -> Option<SharedAsyncSleep> {
 /// Future returned by [`AsyncSleep`].
 #[non_exhaustive]
 #[must_use]
-pub struct Sleep(Pin<Box<dyn Future<Output = ()> + Send + 'static>>);
+pub struct Sleep(Pin<Box<dyn Future<Output = ()> + Send + Sync + 'static>>);
 
 impl Debug for Sleep {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
@@ -95,7 +95,7 @@ impl Sleep {
     /// Create a new [`Sleep`] future
     ///
     /// The provided future will be Boxed.
-    pub fn new(future: impl Future<Output = ()> + Send + 'static) -> Sleep {
+    pub fn new(future: impl Future<Output = ()> + Send + Sync + 'static) -> Sleep {
         Sleep(Box::pin(future))
     }
 }

rust-runtime/aws-smithy-runtime/Cargo.toml

@@ -12,13 +12,13 @@ repository = "https://github.com/awslabs/smithy-rs"
 [features]
 client = ["aws-smithy-runtime-api/client"]
 http-auth = ["aws-smithy-runtime-api/http-auth"]
-connector-hyper-0-14-x = ["dep:hyper", "hyper?/client", "hyper?/http2", "hyper?/http1", "hyper?/tcp"]
+connector-hyper-0-14-x = ["dep:hyper-0-14", "hyper-0-14?/client", "hyper-0-14?/http2", "hyper-0-14?/http1", "hyper-0-14?/tcp", "hyper-0-14?/stream"]
 tls-rustls = ["dep:hyper-rustls", "dep:rustls", "connector-hyper-0-14-x"]
 rt-tokio = ["tokio/rt"]
 
 # Features for testing
 test-util = ["aws-smithy-runtime-api/test-util", "dep:aws-smithy-protocol-test", "dep:tracing-subscriber", "dep:serde", "dep:serde_json"]
-wire-mock = ["test-util", "connector-hyper-0-14-x", "hyper?/server"]
+wire-mock = ["test-util", "connector-hyper-0-14-x", "hyper-0-14?/server"]
 
 [dependencies]
 aws-smithy-async = { path = "../aws-smithy-async" }
@@ -28,9 +28,9 @@ aws-smithy-runtime-api = { path = "../aws-smithy-runtime-api" }
 aws-smithy-types = { path = "../aws-smithy-types", features = ["http-body-0-4-x"] }
 bytes = "1"
 fastrand = "2.0.0"
-http = "0.2.8"
-http-body = "0.4.5"
-hyper = { version = "0.14.26", default-features = false, optional = true }
+http = { version = "0.2.8" }
+http-body-0-4 = { package = "http-body", version = "0.4.4" }
+hyper-0-14 = { package = "hyper", version = "0.14.26", default-features = false, optional = true }
 hyper-rustls = { version = "0.24", features = ["rustls-native-certs", "http2"], optional = true }
 once_cell = "1.18.0"
 pin-project-lite = "0.2.7"
@@ -47,6 +47,8 @@ approx = "0.5.1"
 aws-smithy-async = { path = "../aws-smithy-async", features = ["rt-tokio", "test-util"] }
 aws-smithy-runtime-api = { path = "../aws-smithy-runtime-api", features = ["test-util"] }
 aws-smithy-types = { path = "../aws-smithy-types", features = ["test-util"] }
+futures-util = "0.3.28"
+pretty_assertions = "1.4.0"
 tokio = { version = "1.25", features = ["macros", "rt", "rt-multi-thread", "test-util"] }
 tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
 tracing-test = "0.2.1"

rust-runtime/aws-smithy-runtime/external-types.toml

@@ -25,4 +25,7 @@ allowed_external_types = [
     "hyper::client::connect::Connection",
     "tokio::io::async_read::AsyncRead",
     "tokio::io::async_write::AsyncWrite",
+
+    # TODO(https://github.com/awslabs/smithy-rs/issues/1193): Once tooling permits it, only allow the following types in the `http-0-x` feature
+    "http_body::Body"
 ]

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

@@ -15,3 +15,6 @@ pub mod test_util;
 /// needing to provide equivalent functionality for hyper 1.x in the future.
 #[cfg(feature = "connector-hyper-0-14-x")]
 pub mod hyper_014;
+
+/// HTTP body and body-wrapper types
+pub mod body;

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

@@ -0,0 +1,6 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+pub mod minimum_throughput;

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

@@ -0,0 +1,91 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+//! A body-wrapping type that ensures data is being streamed faster than some lower limit.
+//!
+//! If data is being streamed too slowly, this body type will emit an error next time it's polled.
+
+use aws_smithy_async::rt::sleep::Sleep;
+use aws_smithy_async::rt::sleep::{AsyncSleep, SharedAsyncSleep};
+use aws_smithy_async::time::{SharedTimeSource, TimeSource};
+use aws_smithy_runtime_api::box_error::BoxError;
+use aws_smithy_runtime_api::shared::IntoShared;
+use std::fmt;
+use std::time::Duration;
+use throughput::{Throughput, ThroughputLogs};
+
+/// An implementation of v0.4 `http_body::Body` for `MinimumThroughputBody` and related code.
+pub mod http_body_0_4_x;
+
+mod throughput;
+
+pin_project_lite::pin_project! {
+    /// A body-wrapping type that ensures data is being streamed faster than some lower limit.
+    ///
+    /// If data is being streamed too slowly, this body type will emit an error next time it's polled.
+    pub struct MinimumThroughputBody<B> {
+        async_sleep: SharedAsyncSleep,
+        time_source: SharedTimeSource,
+        minimum_throughput: Throughput,
+        throughput_logs: ThroughputLogs,
+        #[pin]
+        sleep_fut: Option<Sleep>,
+        #[pin]
+        inner: B,
+    }
+}
+
+const SIZE_OF_ONE_LOG: usize = std::mem::size_of::<(std::time::SystemTime, u64)>(); // 24 bytes per log
+const NUMBER_OF_LOGS_IN_ONE_KB: f64 = 1024.0 / SIZE_OF_ONE_LOG as f64;
+
+impl<B> MinimumThroughputBody<B> {
+    /// Create a new minimum throughput body.
+    pub fn new(
+        time_source: impl TimeSource + 'static,
+        async_sleep: impl AsyncSleep + 'static,
+        body: B,
+        (bytes_read, per_time_elapsed): (u64, Duration),
+    ) -> Self {
+        let minimum_throughput = Throughput::new(bytes_read as f64, per_time_elapsed);
+        Self {
+            throughput_logs: ThroughputLogs::new(
+                // Never keep more than 10KB of logs in memory. This currently
+                // equates to 426 logs.
+                (NUMBER_OF_LOGS_IN_ONE_KB * 10.0) as usize,
+                minimum_throughput.per_time_elapsed(),
+            ),
+            async_sleep: async_sleep.into_shared(),
+            time_source: time_source.into_shared(),
+            minimum_throughput,
+            inner: body,
+            sleep_fut: None,
+        }
+    }
+}
+
+#[derive(Debug)]
+enum Error {
+    ThroughputBelowMinimum {
+        expected: Throughput,
+        actual: Throughput,
+    },
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::ThroughputBelowMinimum { expected, actual } => {
+                write!(
+                    f,
+                    "minimum throughput was specified at {expected}, but throughput of {actual} was observed",
+                )
+            }
+        }
+    }
+}
+
+impl std::error::Error for Error {}
+
+// Tests are implemented per HTTP body type.