behavior version 2024-03-28 (#3617)

## Motivation and Context
This PR is the combination of two previously reviewed PRs that both
enable new behaviors behind a new Behavior Major Version (BMV):
* #3527
* #3578

## Description
* Enables stalled stream protection by default on latest behavior
version.
* Enables creation of a default identity cache.

## Testing
All testing done on prior PRs. See for details.

## Checklist
<!--- If a checkbox below is not applicable, then please DELETE it
rather than leaving it unchecked -->
- [x] I have updated `CHANGELOG.next.toml` if I made changes to the
smithy-rs codegen or runtime crates
- [x] I have updated `CHANGELOG.next.toml` if I made changes to the AWS
SDK, generated SDK code, or SDK runtime crates

----

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

---------

Co-authored-by: John DiSanti <jdisanti@amazon.com>
Co-authored-by: ysaito1001 <awsaito@amazon.com>

Author: 3 people

CHANGELOG.next.toml

@@ -10,7 +10,28 @@
 # references = ["smithy-rs#920"]
 # meta = { "breaking" = false, "tada" = false, "bug" = false, "target" = "client | server | all"}
 # author = "rcoh"
+[[aws-sdk-rust]]
+message = """
+`aws-config::loader::ConfigLoader` now creates an `IdentityCache` by default when using `BehaviorVersion::v2024_03_28()`
+or newer. If you're using `BehaviorVersion::latest()`, you will get this change automatically when updating. This
+allows clients created from `SdkConfig` to use the same cache instance by default resulting in fewer cache misses
+when using multiple clients.
+"""
+references = ["smithy-rs#3427"]
+meta = { "breaking" = false, "tada" = true, "bug" = false }
+author = "aajtodd"
 
+[[smithy-rs]]
+message = "Stalled stream protection on uploads is now enabled by default behind `BehaviorVersion::v2024_03_28()`. If you're using `BehaviorVersion::latest()`, you will get this change automatically by running `cargo update`."
+references = ["smithy-rs#3527"]
+meta = { "breaking" = true, "tada" = true, "bug" = false }
+authors = ["jdisanti"]
+
+[[aws-sdk-rust]]
+message = "Stalled stream protection on uploads is now enabled by default behind `BehaviorVersion::v2024_03_28()`. If you're using `BehaviorVersion::latest()`, you will get this change automatically by running `cargo update`. Updating your SDK is not necessary, this change will happen when a new version of the client libraries are consumed."
+references = ["smithy-rs#3527"]
+meta = { "breaking" = true, "tada" = true, "bug" = false }
+author = "jdisanti"
 
 [[smithy-rs]]
 message = "Implement Debug for DateTime"

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

@@ -1,6 +1,6 @@
 [package]
 name = "aws-config"
-version = "1.3.0"
+version = "1.4.0"
 authors = [
     "AWS Rust SDK Team <aws-sdk-rust@amazon.com>",
     "Russell Cohen <rcoh@amazon.com>",

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

@@ -214,6 +214,7 @@ mod loader {
     use aws_credential_types::Credentials;
     use aws_smithy_async::rt::sleep::{default_async_sleep, AsyncSleep, SharedAsyncSleep};
     use aws_smithy_async::time::{SharedTimeSource, TimeSource};
+    use aws_smithy_runtime::client::identity::IdentityCache;
     use aws_smithy_runtime_api::client::behavior_version::BehaviorVersion;
     use aws_smithy_runtime_api::client::http::HttpClient;
     use aws_smithy_runtime_api::client::identity::{ResolveCachedIdentity, SharedIdentityCache};
@@ -238,14 +239,14 @@ mod loader {
     use crate::provider_config::ProviderConfig;
 
     #[derive(Default, Debug)]
-    enum CredentialsProviderOption {
-        /// No provider was set by the user. We can set up the default credentials provider chain.
+    enum TriStateOption<T> {
+        /// No option was set by the user. We can set up the default.
         #[default]
         NotSet,
-        /// The credentials provider was explicitly unset. Do not set up a default chain.
+        /// The option was explicitly unset. Do not set up a default.
         ExplicitlyUnset,
-        /// Use the given credentials provider.
-        Set(SharedCredentialsProvider),
+        /// Use the given user provided option.
+        Set(T),
     }
 
     /// Load a cross-service [`SdkConfig`] from the environment
@@ -258,7 +259,7 @@ mod loader {
     pub struct ConfigLoader {
         app_name: Option<AppName>,
         identity_cache: Option<SharedIdentityCache>,
-        credentials_provider: CredentialsProviderOption,
+        credentials_provider: TriStateOption<SharedCredentialsProvider>,
         token_provider: Option<SharedTokenProvider>,
         endpoint_url: Option<String>,
         region: Option<Box<dyn ProvideRegion>>,
@@ -464,9 +465,8 @@ mod loader {
             mut self,
             credentials_provider: impl ProvideCredentials + 'static,
         ) -> Self {
-            self.credentials_provider = CredentialsProviderOption::Set(
-                SharedCredentialsProvider::new(credentials_provider),
-            );
+            self.credentials_provider =
+                TriStateOption::Set(SharedCredentialsProvider::new(credentials_provider));
             self
         }
 
@@ -492,7 +492,7 @@ mod loader {
         /// # }
         /// ```
         pub fn no_credentials(mut self) -> Self {
-            self.credentials_provider = CredentialsProviderOption::ExplicitlyUnset;
+            self.credentials_provider = TriStateOption::ExplicitlyUnset;
             self
         }
 
@@ -781,14 +781,14 @@ mod loader {
             timeout_config.take_defaults_from(&base_config);
 
             let credentials_provider = match self.credentials_provider {
-                CredentialsProviderOption::Set(provider) => Some(provider),
-                CredentialsProviderOption::NotSet => {
+                TriStateOption::Set(provider) => Some(provider),
+                TriStateOption::NotSet => {
                     let mut builder =
                         credentials::DefaultCredentialsChain::builder().configure(conf.clone());
                     builder.set_region(region.clone());
                     Some(SharedCredentialsProvider::new(builder.build().await))
                 }
-                CredentialsProviderOption::ExplicitlyUnset => None,
+                TriStateOption::ExplicitlyUnset => None,
             };
 
             let token_provider = match self.token_provider {
@@ -851,7 +851,18 @@ mod loader {
             builder.set_behavior_version(self.behavior_version);
             builder.set_http_client(self.http_client);
             builder.set_app_name(app_name);
-            builder.set_identity_cache(self.identity_cache);
+
+            let identity_cache = match self.identity_cache {
+                None => match self.behavior_version {
+                    Some(bv) if bv.is_at_least(BehaviorVersion::v2024_03_28()) => {
+                        Some(IdentityCache::lazy().build())
+                    }
+                    _ => None,
+                },
+                Some(user_cache) => Some(user_cache),
+            };
+
+            builder.set_identity_cache(identity_cache);
             builder.set_credentials_provider(credentials_provider);
             builder.set_token_provider(token_provider);
             builder.set_sleep_impl(sleep_impl);
@@ -1055,10 +1066,26 @@ mod loader {
                 .no_credentials()
                 .load()
                 .await;
-            assert!(config.identity_cache().is_none());
             assert!(config.credentials_provider().is_none());
         }
 
+        #[cfg(feature = "rustls")]
+        #[tokio::test]
+        async fn identity_cache_defaulted() {
+            let config = defaults(BehaviorVersion::latest()).load().await;
+
+            assert!(config.identity_cache().is_some());
+        }
+
+        #[cfg(feature = "rustls")]
+        #[allow(deprecated)]
+        #[tokio::test]
+        async fn identity_cache_old_behavior_version() {
+            let config = defaults(BehaviorVersion::v2023_11_09()).load().await;
+
+            assert!(config.identity_cache().is_none());
+        }
+
         #[tokio::test]
         async fn connector_is_shared() {
             let num_requests = Arc::new(AtomicUsize::new(0));

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/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "aws-types"
-version = "1.2.0"
+version = "1.2.1"
 authors = ["AWS Rust SDK Team <aws-sdk-rust@amazon.com>", "Russell Cohen <rcoh@amazon.com>"]
 description = "Cross-service types for the AWS SDK."
 edition = "2021"

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

@@ -810,9 +810,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/no-default-features/tests/client-construction.rs

@@ -152,6 +152,7 @@ async fn test_time_source_for_identity_cache() {
     let _client = aws_sdk_s3::Client::from_conf(config);
 }
 
+#[allow(deprecated)] // intentionally testing an old behavior version
 #[tokio::test]
 async fn behavior_mv_from_aws_config() {
     let (http_client, req) = capture_request(None);
@@ -177,6 +178,7 @@ async fn behavior_mv_from_aws_config() {
         .starts_with("https://s3.us-west-2.amazonaws.com/"));
 }
 
+#[allow(deprecated)] // intentionally testing an old behavior version
 #[tokio::test]
 async fn behavior_mv_from_client_construction() {
     let (http_client, req) = capture_request(None);

aws/sdk/integration-tests/s3/tests/identity-cache.rs

@@ -6,7 +6,7 @@
 use std::sync::atomic::{AtomicI32, Ordering};
 use std::sync::Arc;
 
-use aws_config::{identity::IdentityCache, BehaviorVersion, Region};
+use aws_config::{BehaviorVersion, Region};
 use aws_credential_types::{
     provider::{future::ProvideCredentials as ProvideCredentialsFuture, ProvideCredentials},
     Credentials,
@@ -23,12 +23,9 @@ async fn test_identity_cache_reused_by_default() {
         infallible_client_fn(|_req| http::Response::builder().status(200).body("OK!").unwrap());
 
     let provider = TestCredProvider::new();
-    let cache = IdentityCache::lazy().build();
     let config = aws_config::defaults(BehaviorVersion::latest())
         .http_client(http_client)
         .credentials_provider(provider.clone())
-        // TODO(rfc-43) - remove adding a cache when this is the new default
-        .identity_cache(cache)
         .region(Region::new("us-west-2"))
         .load()
         .await;
@@ -42,31 +39,7 @@ async fn test_identity_cache_reused_by_default() {
     assert_eq!(1, provider.invoke_count.load(Ordering::SeqCst));
 }
 
-// TODO(rfc-43) - add no_identity_cache() to ConfigLoader and re-enable test
-// #[tokio::test]
-// async fn test_identity_cache_explicit_unset() {
-//     let http_client =
-//         infallible_client_fn(|_req| http::Response::builder().status(200).body("OK!").unwrap());
-//
-//     let provider = TestCredProvider::new();
-//
-//     let config = aws_config::defaults(BehaviorVersion::latest())
-//         .no_identity_cache()
-//         .http_client(http_client)
-//         .credentials_provider(provider.clone())
-//         .region(Region::new("us-west-2"))
-//         .load()
-//         .await;
-//
-//     let c1 = Client::new(&config);
-//     let _ = c1.list_buckets().send().await;
-//     assert_eq!(1, provider.invoke_count.load(Ordering::SeqCst));
-//
-//     let c2 = Client::new(&config);
-//     let _ = c2.list_buckets().send().await;
-//     assert_eq!(2, provider.invoke_count.load(Ordering::SeqCst));
-// }
-
+#[allow(deprecated)] // intentionally testing an old behavior version
 #[tokio::test]
 async fn test_identity_cache_ga_behavior_version() {
     let http_client =

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}"))
@@ -282,12 +282,10 @@ async fn test_stalled_stream_protection_for_downloads_is_enabled_by_default() {
         "minimum throughput was specified at 1 B/s, but throughput of 0 B/s was observed"
     );
     // 5s grace period
-    // TODO(https://github.com/smithy-lang/smithy-rs/issues/3510): Currently comparing against 5 and 6 due to
-    // the behavior change in #3485. Once that feature/fix is released, this should be changed to only check for 5.
     let elapsed_secs = start.elapsed().as_secs();
     assert!(
-        elapsed_secs == 5 || elapsed_secs == 6,
-        "elapsed secs should be 5 or 6, but was {elapsed_secs}"
+        elapsed_secs == 5,
+        "elapsed secs should be 5, but was {elapsed_secs}"
     )
 }