Enable stalled stream protection for uploads behind new behavior version

aws/rust-runtime/aws-inlineable/src/s3_express.rs

@@ -544,7 +544,7 @@ pub(crate) mod identity_provider {
             config_bag: &'a ConfigBag,
         ) -> Result<SessionCredentials, BoxError> {
             let mut config_builder = crate::config::Builder::from_config_bag(config_bag)
-                .behavior_version(self.behavior_version.clone());
+                .behavior_version(self.behavior_version);
 
             // inherits all runtime components from a current S3 operation but clears out
             // out interceptors configured for that operation

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

@@ -798,9 +798,9 @@ impl SdkConfig {
         self.stalled_stream_protection_config.clone()
     }
 
-    /// Behavior major version configured for this client
+    /// Behavior version configured for this client
     pub fn behavior_version(&self) -> Option<BehaviorVersion> {
-        self.behavior_version.clone()
+        self.behavior_version
     }
 
     /// Return an immutable reference to the service config provider configured for this client.

aws/sdk-codegen/src/main/kotlin/software/amazon/smithy/rustsdk/customize/s3/S3ExpressDecorator.kt

@@ -155,7 +155,7 @@ private class S3ExpressServiceRuntimePluginCustomization(codegenContext: ClientC
                             rustTemplate(
                                 """
                                 #{DefaultS3ExpressIdentityProvider}::builder()
-                                    .behavior_version(${section.serviceConfigName}.behavior_version.clone().expect(${behaviorVersionError.dq()}))
+                                    .behavior_version(${section.serviceConfigName}.behavior_version.expect(${behaviorVersionError.dq()}))
                                     .time_source(${section.serviceConfigName}.time_source().unwrap_or_default())
                                     .build()
                                 """,

aws/sdk/integration-tests/s3/tests/stalled-stream-protection.rs

@@ -92,11 +92,10 @@ async fn test_stalled_stream_protection_defaults_for_upload() {
     let _ = tokio::spawn(server);
 
     let conf = Config::builder()
+        // Stalled stream protection MUST BE enabled by default. Do not configure it explicitly.
         .credentials_provider(Credentials::for_tests())
         .region(Region::new("us-east-1"))
         .endpoint_url(format!("http://{server_addr}"))
-        // TODO(https://github.com/smithy-lang/smithy-rs/issues/3510): make stalled stream protection enabled by default with BMV and remove this line
-        .stalled_stream_protection(StalledStreamProtectionConfig::enabled().build())
         .build();
     let client = Client::from_conf(conf);
 
@@ -255,6 +254,7 @@ async fn test_stalled_stream_protection_for_downloads_is_enabled_by_default() {
 
     // Stalled stream protection should be enabled by default.
     let sdk_config = aws_config::from_env()
+        // Stalled stream protection MUST BE enabled by default. Do not configure it explicitly.
         .credentials_provider(Credentials::for_tests())
         .region(Region::new("us-east-1"))
         .endpoint_url(format!("http://{server_addr}"))

codegen-client/src/main/kotlin/software/amazon/smithy/rust/codegen/client/smithy/generators/client/FluentClientGenerator.kt

@@ -493,7 +493,7 @@ private fun baseClientRuntimePluginsFn(
                         .with_client_plugins(#{default_plugins}(
                             #{DefaultPluginParams}::new()
                                 .with_retry_partition_name(${codegenContext.serviceShape.sdkId().dq()})
-                                .with_behavior_version(config.behavior_version.clone().expect(${behaviorVersionError.dq()}))
+                                .with_behavior_version(config.behavior_version.expect(${behaviorVersionError.dq()}))
                         ))
                         // user config
                         .with_client_plugin(

codegen-client/src/main/kotlin/software/amazon/smithy/rust/codegen/client/smithy/generators/config/ServiceConfigGenerator.kt

@@ -438,7 +438,7 @@ class ServiceConfigGenerator(
                         config: self.cloneable.clone(),
                         runtime_components: self.runtime_components.clone(),
                         runtime_plugins: self.runtime_plugins.clone(),
-                        behavior_version: self.behavior_version.clone(),
+                        behavior_version: self.behavior_version,
                     }
                 }
                 """,

rust-runtime/aws-smithy-runtime-api/src/client/behavior_version.rs

@@ -3,17 +3,23 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-//! Behavior Major version of the client
+//! Behavior version of the client
 
-/// Behavior major-version of the client
+/// Behavior version of the client
 ///
 /// Over time, new best-practice behaviors are introduced. However, these behaviors might not be
 /// backwards compatible. For example, a change which introduces new default timeouts or a new
 /// retry-mode for all operations might be the ideal behavior but could break existing applications.
-#[derive(Clone)]
+#[derive(Copy, Clone, PartialEq)]
 pub struct BehaviorVersion {
-    // currently there is only 1 MV so we don't actually need anything in here.
-    _private: (),
+    inner: Inner,
+}
+
+#[derive(Copy, Clone, Debug, Ord, PartialOrd, Eq, PartialEq)]
+enum Inner {
+    // IMPORTANT: Order matters here for the `Ord` derive. Newer versions go to the bottom.
+    V2023_11_09,
+    V2024_03_28,
 }
 
 impl BehaviorVersion {
@@ -26,23 +32,60 @@ impl BehaviorVersion {
     /// If, however, you're writing a service that is very latency sensitive, or that has written
     /// code to tune Rust SDK behaviors, consider pinning to a specific major version.
     ///
-    /// The latest version is currently [`BehaviorVersion::v2023_11_09`]
+    /// The latest version is currently [`BehaviorVersion::v2024_03_28`]
     pub fn latest() -> Self {
-        Self::v2023_11_09()
+        Self::v2024_03_28()
     }
 
-    /// This method returns the behavior configuration for November 9th, 2023
+    /// Behavior version for March 28th, 2024.
+    ///
+    /// This version enables stalled stream protection for uploads (request bodies) by default.
     ///
     /// When a new behavior major version is released, this method will be deprecated.
+    pub fn v2024_03_28() -> Self {
+        Self {
+            inner: Inner::V2024_03_28,
+        }
+    }
+
+    /// Behavior version for November 9th, 2023.
+    #[deprecated(
+        since = "1.4.0",
+        note = "Superceded by v2024_03_28, which enabled stalled stream protection for uploads (request bodies) by default."
+    )]
     pub fn v2023_11_09() -> Self {
-        Self { _private: () }
+        Self {
+            inner: Inner::V2023_11_09,
+        }
+    }
+
+    /// True if this version is newer or equal to the given `other` version.
+    pub fn is_at_least(&self, other: BehaviorVersion) -> bool {
+        self.inner >= other.inner
     }
 }
 
 impl std::fmt::Debug for BehaviorVersion {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("BehaviorVersion")
-            .field("name", &"v2023_11_09")
-            .finish()
+        f.debug_tuple("BehaviorVersion").field(&self.inner).finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    #[allow(deprecated)]
+    fn version_comparison() {
+        assert!(BehaviorVersion::latest() == BehaviorVersion::latest());
+        assert!(BehaviorVersion::v2023_11_09() == BehaviorVersion::v2023_11_09());
+        assert!(BehaviorVersion::v2024_03_28() != BehaviorVersion::v2023_11_09());
+        assert!(BehaviorVersion::latest().is_at_least(BehaviorVersion::latest()));
+        assert!(BehaviorVersion::latest().is_at_least(BehaviorVersion::v2023_11_09()));
+        assert!(BehaviorVersion::latest().is_at_least(BehaviorVersion::v2024_03_28()));
+        assert!(!BehaviorVersion::v2023_11_09().is_at_least(BehaviorVersion::v2024_03_28()));
+        assert!(Inner::V2024_03_28 > Inner::V2023_11_09);
+        assert!(Inner::V2023_11_09 < Inner::V2024_03_28);
     }
 }

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

@@ -176,10 +176,11 @@ pub fn default_identity_cache_plugin() -> Option<SharedRuntimePlugin> {
     note = "This function wasn't intended to be public, and didn't take the behavior major version as an argument, so it couldn't be evolved over time."
 )]
 pub fn default_stalled_stream_protection_config_plugin() -> Option<SharedRuntimePlugin> {
+    #[allow(deprecated)]
     default_stalled_stream_protection_config_plugin_v2(BehaviorVersion::v2023_11_09())
 }
 fn default_stalled_stream_protection_config_plugin_v2(
-    _behavior_version: BehaviorVersion,
+    behavior_version: BehaviorVersion,
 ) -> Option<SharedRuntimePlugin> {
     Some(
         default_plugin(
@@ -191,13 +192,13 @@ fn default_stalled_stream_protection_config_plugin_v2(
             },
         )
         .with_config(layer("default_stalled_stream_protection_config", |layer| {
-            layer.store_put(
-                StalledStreamProtectionConfig::enabled()
-                    // TODO(https://github.com/smithy-lang/smithy-rs/issues/3510): enable behind new behavior version
-                    .upload_enabled(false)
-                    .grace_period(Duration::from_secs(5))
-                    .build(),
-            );
+            let mut config =
+                StalledStreamProtectionConfig::enabled().grace_period(Duration::from_secs(5));
+            // Before v2024_03_28, upload streams did not have stalled stream protection by default
+            if !behavior_version.is_at_least(BehaviorVersion::v2024_03_28()) {
+                config = config.upload_enabled(false);
+            }
+            layer.store_put(config.build());
         }))
         .into_shared(),
     )
@@ -293,3 +294,47 @@ pub fn default_plugins(
     .flatten()
     .collect::<Vec<SharedRuntimePlugin>>()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aws_smithy_runtime_api::client::runtime_plugin::RuntimePlugins;
+
+    fn test_plugin_params(version: BehaviorVersion) -> DefaultPluginParams {
+        DefaultPluginParams::new()
+            .with_behavior_version(version)
+            .with_retry_partition_name("dontcare")
+    }
+    fn config_for(plugins: impl IntoIterator<Item = SharedRuntimePlugin>) -> ConfigBag {
+        let mut config = ConfigBag::base();
+        let plugins = RuntimePlugins::new().with_client_plugins(plugins);
+        plugins.apply_client_configuration(&mut config).unwrap();
+        config
+    }
+
+    #[test]
+    #[allow(deprecated)]
+    fn v2024_03_28_stalled_stream_protection_difference() {
+        let latest = config_for(default_plugins(test_plugin_params(
+            BehaviorVersion::latest(),
+        )));
+        let v2023 = config_for(default_plugins(test_plugin_params(
+            BehaviorVersion::v2023_11_09(),
+        )));
+
+        assert!(
+            latest
+                .load::<StalledStreamProtectionConfig>()
+                .unwrap()
+                .upload_enabled(),
+            "stalled stream protection on uploads MUST be enabled after v2024_03_28"
+        );
+        assert!(
+            !v2023
+                .load::<StalledStreamProtectionConfig>()
+                .unwrap()
+                .upload_enabled(),
+            "stalled stream protection on uploads MUST NOT be enabled before v2024_03_28"
+        );
+    }
+}