Feature: Stalled stream protection

CHANGELOG.next.toml

@@ -177,3 +177,41 @@ The `Unhandled` variant should never be referenced directly.
 references = ["smithy-rs#3191"]
 meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
 author = "jdisanti"
+
+[[aws-sdk-rust]]
+message = """
+Add configurable stalled-stream protection for downloads.
+
+When making HTTP calls,
+it's possible for a connection to 'stall out' and emit no more data due to server-side issues.
+In the event this happens, it's desirable for the stream to error out as quickly as possible.
+While timeouts can protect you from this issue, they aren't adaptive to the amount of data
+being sent and so must be configured specifically for each use case. When enabled, stalled-stream
+protection will ensure that bad streams error out quickly, regardless of the amount of data being
+downloaded.
+
+Protection is enabled by default for all clients but can be configured or disabled.
+See [this discussion](https://github.com/awslabs/aws-sdk-rust/discussions/956) for more details.
+"""
+references = ["smithy-rs#3202"]
+meta = { "breaking" = true, "tada" = true, "bug" = false }
+author = "Velfi"
+
+[[smithy-rs]]
+message = """
+Add configurable stalled-stream protection for downloads.
+
+When making HTTP calls,
+it's possible for a connection to 'stall out' and emit no more data due to server-side issues.
+In the event this happens, it's desirable for the stream to error out as quickly as possible.
+While timeouts can protect you from this issue, they aren't adaptive to the amount of data
+being sent and so must be configured specifically for each use case. When enabled, stalled-stream
+protection will ensure that bad streams error out quickly, regardless of the amount of data being
+downloaded.
+
+Protection is enabled by default for all clients but can be configured or disabled.
+See [this discussion](https://github.com/awslabs/aws-sdk-rust/discussions/956) for more details.
+"""
+references = ["smithy-rs#3202"]
+meta = { "breaking" = true, "tada" = true, "bug" = false, "target" = "client" }
+author = "Velfi"

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

@@ -32,6 +32,7 @@ allowed_external_types = [
    "aws_smithy_types::timeout::OperationTimeoutConfig",
    "aws_smithy_types::timeout::TimeoutConfig",
    "aws_smithy_types::timeout::TimeoutConfigBuilder",
+   "aws_smithy_types::stalled_stream_protection::StalledStreamProtectionConfig",
    "aws_types::*",
    "http::response::Response",
    "http::uri::Uri",

aws/rust-runtime/aws-config/src/lib.rs

@@ -219,6 +219,7 @@ mod loader {
     use aws_smithy_runtime_api::client::identity::{ResolveCachedIdentity, SharedIdentityCache};
     use aws_smithy_runtime_api::shared::IntoShared;
     use aws_smithy_types::retry::RetryConfig;
+    use aws_smithy_types::stalled_stream_protection::StalledStreamProtectionConfig;
     use aws_smithy_types::timeout::TimeoutConfig;
     use aws_types::app_name::AppName;
     use aws_types::docs_for;
@@ -260,6 +261,7 @@ mod loader {
         use_fips: Option<bool>,
         use_dual_stack: Option<bool>,
         time_source: Option<SharedTimeSource>,
+        stalled_stream_protection_config: Option<StalledStreamProtectionConfig>,
         env: Option<Env>,
         fs: Option<Fs>,
         behavior_version: Option<BehaviorVersion>,
@@ -612,6 +614,36 @@ mod loader {
             self
         }
 
+        /// Override [`the StoppedStreamProtectionConfig`] used to build [`SdkConfig`](aws_types::SdkConfig).
+        ///
+        /// This configures stalled stream protection. When enabled, download streams
+        /// that stop (stream no data) for longer than a configured grace period will return an error.
+        ///
+        /// _Note_: When an override is provided, the default implementation is replaced.
+        ///
+        /// # Examples
+        /// ```no_run
+        /// # async fn create_config() {
+        /// use aws_smithy_types::stalled_stream_protection::StalledStreamProtectionConfig;
+        /// use std::time::Duration;
+        /// let config = aws_config::from_env()
+        ///     .stalled_stream_protection_config(
+        ///         StalledStreamProtectionConfig::new_enabled()
+        ///             .grace_period(Duration::from_secs(1))
+        ///             .build()
+        ///     )
+        ///     .load()
+        ///     .await;
+        /// # }
+        /// ```
+        pub fn stalled_stream_protection_config(
+            mut self,
+            stalled_stream_protection_config: StalledStreamProtectionConfig,
+        ) -> Self {
+            self.stalled_stream_protection_config = Some(stalled_stream_protection_config);
+            self
+        }
+
         /// Set configuration for all sub-loaders (credentials, region etc.)
         ///
         /// Update the `ProviderConfig` used for all nested loaders. This can be used to override
@@ -758,6 +790,7 @@ mod loader {
             builder.set_endpoint_url(self.endpoint_url);
             builder.set_use_fips(use_fips);
             builder.set_use_dual_stack(use_dual_stack);
+            builder.set_stalled_stream_protection_config(self.stalled_stream_protection_config);
             builder.build()
         }
     }

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

@@ -17,4 +17,5 @@ allowed_external_types = [
     "aws_smithy_types::error::metadata::Builder",
     "aws_smithy_types::retry::RetryConfig",
     "aws_smithy_types::timeout::TimeoutConfig",
+    "aws_smithy_types::stalled_stream_protection::StalledStreamProtectionConfig",
 ]

aws/rust-runtime/aws-types/src/sdk_config.rs

@@ -23,6 +23,7 @@ pub use aws_smithy_runtime_api::client::http::SharedHttpClient;
 use aws_smithy_runtime_api::client::identity::{ResolveCachedIdentity, SharedIdentityCache};
 use aws_smithy_runtime_api::shared::IntoShared;
 pub use aws_smithy_types::retry::RetryConfig;
+pub use aws_smithy_types::stalled_stream_protection::StalledStreamProtectionConfig;
 pub use aws_smithy_types::timeout::TimeoutConfig;
 
 #[doc(hidden)]
@@ -60,6 +61,7 @@ pub struct SdkConfig {
     sleep_impl: Option<SharedAsyncSleep>,
     time_source: Option<SharedTimeSource>,
     timeout_config: Option<TimeoutConfig>,
+    stalled_stream_protection_config: Option<StalledStreamProtectionConfig>,
     http_client: Option<SharedHttpClient>,
     use_fips: Option<bool>,
     use_dual_stack: Option<bool>,
@@ -82,6 +84,7 @@ pub struct Builder {
     sleep_impl: Option<SharedAsyncSleep>,
     time_source: Option<SharedTimeSource>,
     timeout_config: Option<TimeoutConfig>,
+    stalled_stream_protection_config: Option<StalledStreamProtectionConfig>,
     http_client: Option<SharedHttpClient>,
     use_fips: Option<bool>,
     use_dual_stack: Option<bool>,
@@ -567,10 +570,79 @@ impl Builder {
             use_dual_stack: self.use_dual_stack,
             time_source: self.time_source,
             behavior_version: self.behavior_version,
+            stalled_stream_protection_config: self.stalled_stream_protection_config,
         }
     }
 }
 
+impl Builder {
+    /// Set the [`StalledStreamProtectionConfig`] to configure protection for stalled streams.
+    ///
+    /// This configures stalled stream protection. When enabled, download streams
+    /// that stall (stream no data) for longer than a configured grace period will return an error.
+    ///
+    /// _Note:_ Stalled stream protection requires both a sleep implementation and a time source
+    /// in order to work. When enabling stalled stream protection, make sure to set
+    /// - A sleep impl with [Self::sleep_impl] or [Self::set_sleep_impl].
+    /// - A time source with [Self::time_source] or [Self::set_time_source].
+    ///
+    /// # Examples
+    /// ```rust
+    /// use std::time::Duration;
+    /// use aws_types::SdkConfig;
+    /// pub use aws_smithy_types::stalled_stream_protection::StalledStreamProtectionConfig;
+    ///
+    /// let stalled_stream_protection_config = StalledStreamProtectionConfig::new_enabled()
+    ///     .grace_period(Duration::from_secs(1))
+    ///     .build();
+    /// let config = SdkConfig::builder()
+    ///     .stalled_stream_protection_config(stalled_stream_protection_config)
+    ///     .build();
+    /// ```
+    pub fn stalled_stream_protection_config(
+        mut self,
+        stalled_stream_protection_config: StalledStreamProtectionConfig,
+    ) -> Self {
+        self.set_stalled_stream_protection_config(Some(stalled_stream_protection_config));
+        self
+    }
+
+    /// Set the [`StalledStreamProtectionConfig`] to configure protection for stalled streams.
+    ///
+    /// This configures stalled stream protection. When enabled, download streams
+    /// that stall (stream no data) for longer than a configured grace period will return an error.
+    ///
+    /// _Note:_ Stalled stream protection requires both a sleep implementation and a time source
+    /// in order to work. When enabling stalled stream protection, make sure to set
+    /// - A sleep impl with [Self::sleep_impl] or [Self::set_sleep_impl].
+    /// - A time source with [Self::time_source] or [Self::set_time_source].
+    ///
+    /// # Examples
+    /// ```rust
+    /// use std::time::Duration;
+    /// use aws_types::sdk_config::{SdkConfig, Builder};
+    /// pub use aws_smithy_types::stalled_stream_protection::StalledStreamProtectionConfig;
+    ///
+    /// fn set_stalled_stream_protection(builder: &mut Builder) {
+    ///     let stalled_stream_protection_config = StalledStreamProtectionConfig::new_enabled()
+    ///         .grace_period(Duration::from_secs(1))
+    ///         .build();
+    ///     builder.set_stalled_stream_protection_config(Some(stalled_stream_protection_config));
+    /// }
+    ///
+    /// let mut builder = SdkConfig::builder();
+    /// set_stalled_stream_protection(&mut builder);
+    /// let config = builder.build();
+    /// ```
+    pub fn set_stalled_stream_protection_config(
+        &mut self,
+        stalled_stream_protection_config: Option<StalledStreamProtectionConfig>,
+    ) -> &mut Self {
+        self.stalled_stream_protection_config = stalled_stream_protection_config;
+        self
+    }
+}
+
 impl SdkConfig {
     /// Configured region
     pub fn region(&self) -> Option<&Region> {
@@ -633,6 +705,11 @@ impl SdkConfig {
         self.use_dual_stack
     }
 
+    /// Configured stalled stream protection
+    pub fn stalled_stream_protection_config(&self) -> Option<StalledStreamProtectionConfig> {
+        self.stalled_stream_protection_config.clone()
+    }
+
     /// Behavior major version configured for this client
     pub fn behavior_version(&self) -> Option<BehaviorVersion> {
         self.behavior_version.clone()
@@ -668,6 +745,7 @@ impl SdkConfig {
             use_fips: self.use_fips,
             use_dual_stack: self.use_dual_stack,
             behavior_version: self.behavior_version,
+            stalled_stream_protection_config: self.stalled_stream_protection_config,
         }
     }
 }

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

@@ -21,7 +21,7 @@ aws-credential-types = { path = "../../build/aws-sdk/sdk/aws-credential-types",
 aws-http = { path = "../../build/aws-sdk/sdk/aws-http" }
 aws-runtime = { path = "../../build/aws-sdk/sdk/aws-runtime", features = ["test-util"] }
 aws-sdk-s3 = { path = "../../build/aws-sdk/sdk/s3", features = ["test-util", "behavior-version-latest"] }
-# aws-sdk-sts = { path = "../../build/aws-sdk/sdk/sts" }
+aws-sdk-sts = { path = "../../build/aws-sdk/sdk/sts" }
 aws-smithy-async = { path = "../../build/aws-sdk/sdk/aws-smithy-async", features = ["test-util", "rt-tokio"] }
 aws-smithy-http = { path = "../../build/aws-sdk/sdk/aws-smithy-http" }
 aws-smithy-protocol-test = { path = "../../build/aws-sdk/sdk/aws-smithy-protocol-test" }