Thread through BehaviorVersion to S3 Express identity provider (#3478)

## Motivation and Context
A follow-up task that resolves one of the `TODO(Post S3Express release)`

## Description
This PR ensures that `BehaviorVersion` is threaded through from an outer
S3 client to the inner S3 client when `DefaultS3ExpressIdentityProvider`
is constructed.

I considered the alternative where `BehaviorVersion` would be stored in
`ConfigBag`using another flavor like `StoreOnce` and obtained from the
bag within
`DefaultS3ExpressIdentityProvider::express_session_credentials`. But
ensuring `StoreOnce` per `ConfigBag` was not enough for
`BehaviorVersion` since there could be nested `ConfigBag`s where each
`ConfigBag` was separate. `BehaviorVersion` needs to be unique across
those different `ConfigBag`s and the fact that we store
`BehaviorVersion` outside `RuntimeComponents` or `ConfigBag` is a
reasonable choice from that perspective (the ease of guaranteeing the
uniqueness of `BehaviorVersion`).

## Testing
Relied on existing tests in CI.

## 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 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._

Author: ysaito1001

CHANGELOG.next.toml

@@ -16,3 +16,9 @@ message = "Increased minimum version of wasi crate dependency in aws-smithy-wasm
 references = ["smithy-rs#3476"]
 meta = { "breaking" = false, "tada" = false, "bug" = false }
 authors = ["landonxjames"]
+
+[[aws-sdk-rust]]
+message = "`DefaultS3ExpressIdentityProvider` now uses `BehaviorVersion` threaded through from the outer S3 client, instead of always creating `BehaviorVersion::latest()` on the fly."
+references = ["smithy-rs#3478"]
+meta = { "breaking" = false, "bug" = true, "tada" = false }
+author = "ysaito1001"

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

@@ -464,6 +464,7 @@ pub(crate) mod identity_provider {
 
     #[derive(Debug)]
     pub(crate) struct DefaultS3ExpressIdentityProvider {
+        behavior_version: crate::config::BehaviorVersion,
         cache: S3ExpressIdentityCache,
     }
 
@@ -542,9 +543,8 @@ pub(crate) mod identity_provider {
             runtime_components: &'a RuntimeComponents,
             config_bag: &'a ConfigBag,
         ) -> Result<SessionCredentials, BoxError> {
-            // TODO(Post S3Express release): Thread through `BehaviorVersion` from the outer S3 client
             let mut config_builder = crate::config::Builder::from_config_bag(config_bag)
-                .behavior_version(crate::config::BehaviorVersion::latest());
+                .behavior_version(self.behavior_version.clone());
 
             // inherits all runtime components from a current S3 operation but clears out
             // out interceptors configured for that operation
@@ -568,11 +568,26 @@ pub(crate) mod identity_provider {
 
     #[derive(Default)]
     pub(crate) struct Builder {
+        behavior_version: Option<crate::config::BehaviorVersion>,
         time_source: Option<SharedTimeSource>,
         buffer_time: Option<Duration>,
     }
 
     impl Builder {
+        pub(crate) fn behavior_version(
+            mut self,
+            behavior_version: crate::config::BehaviorVersion,
+        ) -> Self {
+            self.set_behavior_version(Some(behavior_version));
+            self
+        }
+        pub(crate) fn set_behavior_version(
+            &mut self,
+            behavior_version: Option<crate::config::BehaviorVersion>,
+        ) -> &mut Self {
+            self.behavior_version = behavior_version;
+            self
+        }
         pub(crate) fn time_source(mut self, time_source: impl TimeSource + 'static) -> Self {
             self.set_time_source(time_source.into_shared());
             self
@@ -593,6 +608,9 @@ pub(crate) mod identity_provider {
         }
         pub(crate) fn build(self) -> DefaultS3ExpressIdentityProvider {
             DefaultS3ExpressIdentityProvider {
+                behavior_version: self
+                    .behavior_version
+                    .expect("required field `behavior_version` should be set"),
                 cache: S3ExpressIdentityCache::new(
                     DEFAULT_MAX_CACHE_CAPACITY,
                     self.time_source.unwrap_or_default(),

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

@@ -28,6 +28,7 @@ import software.amazon.smithy.rust.codegen.core.rustlang.writable
 import software.amazon.smithy.rust.codegen.core.smithy.RuntimeConfig
 import software.amazon.smithy.rust.codegen.core.smithy.RuntimeType
 import software.amazon.smithy.rust.codegen.core.smithy.RuntimeType.Companion.preludeScope
+import software.amazon.smithy.rust.codegen.core.util.dq
 import software.amazon.smithy.rust.codegen.core.util.getTrait
 import software.amazon.smithy.rustsdk.AwsCargoDependency
 import software.amazon.smithy.rustsdk.AwsRuntimeType
@@ -105,6 +106,9 @@ private class S3ExpressServiceRuntimePluginCustomization(codegenContext: ClientC
                     .resolve("client::identity::SharedIdentityResolver"),
         )
     }
+    val behaviorVersionError =
+        "Invalid client configuration: A behavior version must be set when creating an inner S3 client. " +
+            "A behavior version should be set in the outer S3 client, so it needs to be passed down to the inner client."
 
     override fun section(section: ServiceRuntimePluginSection): Writable =
         writable {
@@ -126,6 +130,7 @@ private class S3ExpressServiceRuntimePluginCustomization(codegenContext: ClientC
                             rustTemplate(
                                 """
                                 #{DefaultS3ExpressIdentityProvider}::builder()
+                                    .behavior_version(${section.serviceConfigName}.behavior_version.clone().expect(${behaviorVersionError.dq()}))
                                     .time_source(${section.serviceConfigName}.time_source().unwrap_or_default())
                                     .build()
                                 """,

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

@@ -486,16 +486,14 @@ private fun baseClientRuntimePluginsFn(
                 ) -> #{RuntimePlugins} {
                     let mut configured_plugins = #{Vec}::new();
                     ::std::mem::swap(&mut config.runtime_plugins, &mut configured_plugins);
-                    ##[allow(unused_mut)]
-                    let mut behavior_version = config.behavior_version.clone();
                     #{update_bmv}
 
                     let mut plugins = #{RuntimePlugins}::new()
                         // defaults
                         .with_client_plugins(#{default_plugins}(
                             #{DefaultPluginParams}::new()
                                 .with_retry_partition_name(${codegenContext.serviceShape.sdkId().dq()})
-                                .with_behavior_version(behavior_version.expect(${behaviorVersionError.dq()}))
+                                .with_behavior_version(config.behavior_version.clone().expect(${behaviorVersionError.dq()}))
                         ))
                         // user config
                         .with_client_plugin(
@@ -532,8 +530,8 @@ private fun baseClientRuntimePluginsFn(
                     featureGatedBlock(BehaviorVersionLatest) {
                         rustTemplate(
                             """
-                            if behavior_version.is_none() {
-                                behavior_version = Some(#{BehaviorVersion}::latest());
+                            if config.behavior_version.is_none() {
+                                config.behavior_version = Some(#{BehaviorVersion}::latest());
                             }
 
                             """,