Client Orchestrator Upgrade Guidance

[jdisanti]

Generated smithy-rs clients have undergone an architectural overhaul that replaces the tower Service trait (referred to as the “middleware” implementation) with a new orchestration system that will be referred to as the “orchestrator” for the rest of this post. Users that implement middleware via tower Service or any of our MapRequest layers will experience significant breaking changes that will require refactoring to resolve.

This new implementation brings a much easier client creation API, and a less error-prone API for customizing the client that has more extension points and encourages best practices. It also brings smithy-rs generated clients closer to how generated clients work in other code generators (e.g., smithy-kotlin, smithy-swift, and others in the future) so that customizations written for one language’s Smithy clients can be easily ported to another language. In the long term, the overall high-level client architecture will be the same across the board for Smithy clients in all languages.

At the highest level, the orchestrator can be described by its parts:

• Config is used to configure the client. This hasn’t really changed much from the middleware implementation.
• Runtime Components are configurable implementations that the orchestrator will use to resolve endpoints, identities and auth schemes, make requests, and more.
• Runtime Plugins can be registered in the config to change runtime components and config.
• Interceptors are hook points into the request lifecycle that can read and mutate the input, request, response, and output. These are registered in config.

When a request is made, the config is augmented by the runtime plugins, and the runtime components for that request are established by the runtime plugins. The list of interceptors is retrieved from config, and then the orchestrator goes through a fixed request execution pipeline with various hook points for customization with interceptors and runtime components.

There isn’t a ton of documentation on how this all works under the hood yet, but we do have the following docs, as well as the docs of the aws-smithy-runtime-api and aws-smithy-runtime crates, which cover the finer details.

• RFC-0034: Smithy Orchestrator
• What is the ‘orchestrator’ and why does it exist?
• Identity and Auth in Clients
• aws-smithy-runtime-api docs (note: it may take a few days for docs.rs to update these after release)
• aws-smithy-runtime docs

Middleware deprecation

The release of smithy-rs v0.56.0 makes the new orchestrator the default client implementation. The middleware implementation is still around and can be opted into if necessary, but this his highly discouraged as it will be completely removed and unsupported in the next minor (non-patch) smithy-rs release. If you find that your app is not behaving as expected after the upgrade, you have a few options:

• Submit a bug report or feature request. This is the best way to help improve the new orchestrator-based client and we’d like to hear what you have to say. Get in touch with us early so that we can make sure the next release meets your requirements.
• Temporarily opt out of the orchestrator and back into middleware. Instructions on how to do this are at the very end of this doc.
• Downgrade to an older version. We’ve done our best to ensure that this isn’t necessary, but nobody’s perfect. If you’ve already tried the two approaches listed above, then you can still downgrade to v0.55.0.

Interceptors aren’t middleware

But they’re pretty close. Here are the key differences:

• Interceptors don’t require implementers to define complex generic bounds.
• Interceptors can’t redirect control flow in the same ways that middlewares can. While interceptors can return errors, they can’t stop other interceptors from running. Nor can an interceptor return a response early – i.e. middlewares that can “short-circuit” a call with a success response won’t be possible to implement as an interceptor.

Interceptors are a generic extension point with the goal of supporting “anything someone should be able to do” as opposed to “anything anyone might want to do.” Interceptors allow users to “inject” logic into specific stages of the request/response lifecycle. These stages are known as “hooks”.

Hooks are either read-only (read) or read/write (modify). The current list of hooks is as follows:

1. Read Before Execution - Called before anything happens. This is the first thing the SDK calls during operation execution.
2. Modify Before Serialization - Called before the input given by the user is serialized into an HTTP request. Allows modifying the input.
3. Read Before Serialization - The last thing the orchestrator calls before serializing the input into an HTTP request.
4. Read After Serialization - The first thing the orchestrator calls after serializing the input into an HTTP request.
5. (Retry Loop)
   1. Modify Before Retry Loop - The last thing the orchestrator calls before entering the retry loop. Allows modifying the HTTP request.
   2. Read Before Attempt - The first thing the orchestrator calls “inside” of the retry loop.
   3. Modify Before Signing - Before the HTTP request is signed. Allows modifying the HTTP request.
   4. Read Before Signing - The last thing the orchestrator calls before signing the HTTP request.
   5. Read After Signing - The first thing the orchestrator calls after signing the HTTP request.
   6. Modify Before Transmit - Before the HTTP request is sent to the service. Allows modifying the HTTP request.
   7. Read Before Transmit - The last thing the orchestrator calls before sending the HTTP request.
   8. Read After Transmit - The last thing the orchestrator calls after receiving the HTTP response.
   9. Modify Before Deserialization - Before the HTTP response is deserialized. Allows modifying the HTTP response.
   10. Read Before Deserialization - The last thing the orchestrator calls before deserializing the HTTP response into an output.
   11. Read After Deserialization - The last thing the orchestrator calls after deserializing the HTTP response into an output.
   12. Modify Before Attempt Completion - Before the retry loop ends. Allows modifying the output.
   13. Read After Attempt - The last thing the orchestrator calls “inside” of the retry loop.
6. Modify Before Execution Completion - Before the execution ends. Allows modifying the output.
7. Read After Execution - The final hook in the request/response lifecycle.

A hook will always run all applicable interceptors, even if one fails. If one or more interceptors run by a hook fail, orchestrator execution will skip ahead. The hook that is skipped to depends on whether the retry loop was entered. The Modify Before Attempt Completion and Read After Attempt hooks will always run if an interceptor fails during the retry loop. The Modify Before Execution Completion and Read After Execution hooks will always run.

If you’re relying on custom middlewares, you’ll need to convert them to the interceptor pattern in order to upgrade the SDK. Let’s look at two of the middlewares the SDK team had to convert as an example.

What hook(s) should I implement for my interceptor?

While it’s possible for a single interceptor to implement all of the above hooks at once, implementing one is enough for most use-cases. The question, then, is “which one should I implement?” When making this decision, here are a few things to consider:

1. Do you need to access the interceptor context (input, request, response, or output)?
   1. Your interceptor will have to run in a hook where the necessary context exists. For example, You can’t access the HTTP request until after it’s been serialized by the orchestrator. The Read After Serialization hook is the earliest time that that’s possible.
2. Do you need to modify the interceptor context or just read it?
   1. Modifying the context (or the config bag) can only be done in a Modify_* hook.
3. Should your interceptor run once for each attempt (including retries) or just once for all attempts?
   1. Hooks occurring outside of the retry loop will only be run once. Hooks inside the retry loop will run once for each attempt (the initial request as well as retry attempts.)
4. Do you want the interceptor to always run, no matter what?
   1. Interceptors that must always run should implement the Modify Before Execution Completion or Read After Execution hook. If the retry loop is entered, then the Modify Before Attempt Completion and Read After Attempt hooks will be run; However, if the orchestrator encounters an error before entering the retry loop, then it will skip to the Modify Before Execution Completion hook, so they may not always run.
5. If you need to grab a value from the ConfigBag, when is that value set?
   1. Your interceptor will have to run in the next hook after the value is set.
6. Does your interceptor insert headers into the request?
   1. Implement the Modify Before Signing hook (or an earlier hook) so that the inserted headers will be included when signing the request. Otherwise, the service will respond with a signature error.

NOTE: The most common hook implemented by our internal interceptors is Modify Before Signing. The second most common hook is Modify Before Transmit.

What if my middleware follows redirects, or modifies/replaces the request body?

Middleware that follow redirects or send multiple HTTP requests also don't translate to interceptors.

Middleware that completely replace the HTTP request body will not translate to interceptors well. You can use an interceptor to std::mem::swap the HTTP body out with another one (and SdkBody can wrap any arbitrary streaming body implementation), but unless you replace the body with an in-memory variant of SdkBody, you will lose the ability to retry. Note: you would have potentially lost the ability to retry in middleware also, depending on how the middleware was implemented.

If you need to follow redirects or replace the HTTP body with one that is streaming as opposed to in-memory, you should do it as a custom HttpConnector instead of trying to do it in an interceptor.

Setting interceptors

All necessary interceptors are set by default, but users may want to define and set their own. Interceptors can be added one-at-a-time or in a list:

#[tokio::main]
async fn main() {
    let sdk_config = aws_config::load_from_env();

    let conf = aws_sdk_s3::config::Builder::from(&sdk_config)
        // Option #1:
        // Add interceptors by repeatedly calling the ``interceptor`` method
        .interceptor(YourInterceptor::new())
        .build();
    let client = aws_sdk_s3::Client::from_conf(conf);
}

The above method works by taking ownership of self. The two methods below take a mutable reference to self.

#[tokio::main]
async fn main() {
    let sdk_config = aws_config::load_from_env();

    let mut conf_builder = aws_sdk_s3::config::Builder::from(&sdk_config);
     // Option #2:
    // Add interceptors by repeatedly calling the `add_interceptor` method.
    // You'll need to wrap each one in a `SharedInterceptor`.
    conf_builder.push_interceptor(SharedInterceptor::new(YourInterceptor::new()));
    // Option #3:
    // Set all of your interceptors at once by passing in an iterator/vec.
    // You'll need to wrap each one in a `SharedInterceptor`.
    // **This will unset interceptors set using methods #1 or #2**
    conf_builder.set_interceptors(vec![
        SharedInterceptor::new(YourInterceptor::new()),
        SharedInterceptor::new(YourInterceptor::new()),
    ]);
    let conf = conf_builder.build();
    let client = aws_sdk_s3::Client::from_conf(conf);
}

Example #1: Updating the recursion detection middleware

The recursion detection middleware is simple, but has a very important job. It detects when an SDK is being run in a Lambda and inserts a special tracking header so that services handling the request can avoid infinitely recursive lambda invocations.

Both the middleware and interceptors implementations rely on the same inner function:

fn augument_request<B>(req: &mut http::Request<B>, env: &Env) {
    // Do nothing if a trace header was already set
    if req.headers().contains_key(TRACE_ID_HEADER) {
        return;
    }
     // Check to see if we're running in a lambda and then insert the trace ID
    // into the request.
    if let (Ok(_function_name), Ok(trace_id)) =
        (env.get(env::LAMBDA_FUNCTION_NAME), env.get(TRACE_ID))
    {
        req.headers_mut()
            .insert(TRACE_ID_HEADER, encode_header(trace_id.as_bytes()));
    }
}

The middleware is defined as a struct that implements the MapRequest helper trait:

#[derive(Default, Debug, Clone)]
pub struct RecursionDetectionStage {
    env: Env,
}

impl RecursionDetectionStage {
    /// Creates a new `RecursionDetectionStage`
    pub fn new() -> Self {
        Self::default()
    }
}

impl MapRequest for RecursionDetectionStage {
    type Error = std::convert::Infallible;

    fn name(&self) -> &ˈstatic str {
        "recursion_detection"
    }

    fn apply(&self, request: Request) -> Result<Request, Self::Error> {
        request.augment(|mut req, _conf| {
            augument_request(&mut req, &self.env);
            Ok(req)
        })
    }
}

When converted to the new interceptor pattern, it looks like this:

#[derive(Debug, Default)]
pub struct RecursionDetectionInterceptor {
    env: Env,
}

impl RecursionDetectionInterceptor {
    /// Creates a new `RecursionDetectionInterceptor`
    pub fn new() -> Self {
        Self::default()
    }
}

impl Interceptor for RecursionDetectionInterceptor {
    fn name(&self) -> &'static str {
        "RecursionDetectionInterceptor"
    }

    fn modify_before_signing(
        &self,
        context: &mut BeforeTransmitInterceptorContextMut<'_>,
        _runtime_components: &RuntimeComponents,
        _cfg: &mut ConfigBag,
    ) -> Result<(), BoxError> {
        let request = context.request_mut();
        augment_request(request, &self.env);
        Ok(())
    }
}

Not so different, right? The most significant decision when converting simple middleware into interceptors is what hook(s) to implement. In this case, we chose the modify_before_signing hook, inserting the trace ID header right before signing the request. That way, the configured signer may choose to sign it.

Note: all the types needed to implement an interceptor are re-exported in crate::config, and crate::config::interceptors.

Example #2: Reconnecting when retrying 50X errors

By default, the AWS SDK for Rust won’t re-use a connection if it’s retrying a server error. This helps avoid sending another request to a server that’s struggling to respond, instead allowing another server to handle it. The way this is implemented is a bit complex but that’s what makes it a good example.

The PoisonService middleware is responsible for “poisoning” connections to servers that have returned 50X errors. When a connection is “poisoned” that signals to the underlying HTTP client that it shouldn’t be re-used for future requests.

Because the PoisonService is a bit more complex, it implements the tower::Service trait instead of the MapRequest helper trait:

pub(crate) struct PoisonLayer<S> {
    inner: PhantomData<S>,
    mode: ReconnectMode,
}

impl<S> PoisonLayer<S> {
    pub(crate) fn new(mode: ReconnectMode) -> Self {
        Self {
            inner: Default::default(),
            mode,
        }
    }
}

impl<S> Clone for PoisonLayer<S> {
    fn clone(&self) -> Self {
        Self {
            inner: Default::default(),
            mode: self.mode,
        }
    }
}

impl<S> tower::Layer<S> for PoisonLayer<S> {
    type Service = PoisonService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        PoisonService {
            inner,
            mode: self.mode,
        }
    }
}

#[derive(Clone)]
pub(crate) struct PoisonService<S> {
    inner: S,
    mode: ReconnectMode,
}

impl<H, R, S, O, E> tower::Service<Operation<H, R>> for PoisonService<S>
where
    R: ClassifyRetry<SdkSuccess<O>, SdkError<E>>,
    S: tower::Service<Operation<H, R>, Response = SdkSuccess<O>, Error = SdkError<E>>,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = PoisonServiceFuture<S::Future, R>;

    fn poll_ready(&mut self, cx: &mut Context<ˈ_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: Operation<H, R>) -> Self::Future {
        let classifier = req.retry_classifier().clone();
        // Create a special struct that will "capture" a connection when a request
        // is set. If the response that comes back is a 50X error, we'll call the 
        // .poison() method on the captured connection before retrying the request.
        let capture_smithy_connection = CaptureSmithyConnection::new();
        // The HTTP Client which manages connections will react to seeing this
        // in the property bag by inserting a reference to the connection into
        // the capture_smithy_connection struct.
        req.properties_mut()
            .insert(capture_smithy_connection.clone());
        PoisonServiceFuture {
            inner: self.inner.call(req),
            conn: capture_smithy_connection,
            mode: self.mode,
            classifier,
        }
    }
}

pin_project! {
    pub struct PoisonServiceFuture<F, R> {
        #[pin]
        inner: F,
        classifier: R,
        conn: CaptureSmithyConnection,
        mode: ReconnectMode
    }
}

impl<F, R, T, E> Future for PoisonServiceFuture<F, R>
where
    F: Future<Output = Result<SdkSuccess<T>, SdkError<E>>>,
    R: ClassifyRetry<SdkSuccess<T>, SdkError<E>>,
{
    type Output = F::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<ˈ_>) -> Poll<Self::Output> {
        let this = self.project();
        match this.inner.poll(cx) {
            Poll::Ready(resp) => {
                let retry_kind = this.classifier.classify_retry(resp.as_ref());
                // This is where we check the response and poison the connection
                // if necessary.
                if this.mode == &ReconnectMode::ReconnectOnTransientError
                    && retry_kind == RetryKind::Error(ErrorKind::TransientError)
                {
                    if let Some(smithy_conn) = this.conn.get() {
                        tracing::info!("poisoning connection: {:?}", smithy_conn);
                        smithy_conn.poison();
                    } else {
                        tracing::trace!("No smithy connection found! The underlying HTTP connection never set a connection.");
                    }
                }
                Poll::Ready(resp)
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

When converted to an interceptor, we end up with this:

#[derive(Debug, Default)]
pub struct ConnectionPoisoningInterceptor {}

impl ConnectionPoisoningInterceptor {
    pub fn new() -> Self {
        Self::default()
    }
}

impl Interceptor for ConnectionPoisoningInterceptor {
    fn name(&self) -> &'static str {
        "ConnectionPoisoningInterceptor"
    }

    fn modify_before_transmit(
        &self,
        context: &mut BeforeTransmitInterceptorContextMut<ˈ_>,
        cfg: &mut ConfigBag,
    ) -> Result<(), BoxError> {
    
        let capture_smithy_connection = CaptureSmithyConnectionWrapper::new();
        // In this case, the HTTP client can't see into interceptor state, so we
        // have to store the capture_smithy_connection struct into the request's
        // .extensions() property bag.
        context
            .request_mut()
            .extensions_mut()
            .insert(capture_smithy_connection.clone_inner());
        cfg.interceptor_state().store_put(capture_smithy_connection);

        Ok(())
    }

    fn modify_before_deserialization(
        &self,
        context: &mut BeforeDeserializationInterceptorContextMut<ˈ_>,
        _runtime_components: &RuntimeComponents,
        cfg: &mut ConfigBag,
    ) -> Result<(), BoxError> {
        let reconnect_mode = cfg
            .load::<RetryConfig>()
            .map(RetryConfig::reconnect_mode)
            .unwrap_or(ReconnectMode::ReconnectOnTransientError);
        let captured_connection = cfg.load::<CaptureSmithyConnectionWrapper>().cloned();
        let retry_classifiers = cfg.retry_classifiers();

        let error_is_transient = retry_classifiers
            .classify_retry(context.into_inner())
            .map(|reason| reason == RetryReason::Error(ErrorKind::TransientError))
            .unwrap_or_default();
        let connection_poisoning_is_enabled =
            reconnect_mode == ReconnectMode::ReconnectOnTransientError;

        if error_is_transient && connection_poisoning_is_enabled {
            debug!("received a transient error, poisoning the connection...");

            // After we get a response and it's a 50X error, we poison the captured
            // connection.
            if let Some(captured_connection) = captured_connection.and_then(|conn| conn.get()) {
                captured_connection.poison();
                debug!("the connection was poisoned")
            } else {
                error!(
                    "unable to poison the connection because no connection was found! The underlying HTTP connector never set a connection."
                );
            }
        }

        Ok(())
    }
}

Summary

We hope this new method of extending the request/response lifecycle will be easy to use, but if you have trouble upgrading your middlewares, please file an issue and we’ll do our best to advise you.

Config hasn’t changed much

While the internals of SDK configuration have changed, we’ve put effort into ensuring the user experience remains the same. SdkConfig and service specific configs will have all the same fields they did before, as well as a few additions:

• push_interceptor: Add a SharedInterceptor that runs for specific hooks in the request/response lifecycle. Generally speaking, interceptors set to run on a hook are executed according to the pre-defined priority in a non-deterministic order. However, the default set of interceptors provided by the SDK are guaranteed to run before interceptors set by the user, ensuring the user always has the final say.
  • A SharedInterceptor is a wrapper type for structs implementing Interceptor.
• time_source: Set the time source for a client. The time source is used to provide the current time. By default, this will use the host machine’s system time. If you’re running on WASM, you’ll have to set your own time source depending on your specific runtime.
  • Setting a custom time source is useful for some kinds of tests.

Generic clients are simpler to use

If you’re generating a generic smithy client, client configuration and construction works just like it does for SDK clients, but with less defaults:

#[tokio::main]
async fn main () {
    let config = generic_smithy_service::Config::builder()
        // To resolve a specific endpoint, pass a `&'str`.
        .endpoint_resolver("https://localhost:13734")
        // A sleep implementation is needed for retries and timeouts.
        .sleep_impl(SharedAsyncSleep::new(default_async_sleep().unwrap()))
        // A source of time is also needed.
        .time_source(SharedTimeSource::new(SystemTimeSource::new()))
        // Retry configuration works the same for generic and SDK clients
        .retry_config(RetryConfig::standard())
        // Timeout configuration works the same for generic and SDK clients
        .timeout_config(TimeoutConfig::builder()
            .operation_timeout(Duration::from_secs(30))
            .operation_attempt_timeout(Duration::from_secs(10))
            .connect_timeout(Duration::from_secs(3))
            .read_timeout(Duration::from_secs(3))
            .build()
        )
        .build();
    // Clients are created from a Config.
    let client = generic_smithy_service::client::Client::from_conf(config);

    let res = client.get_something()
        // Operation parameter methods would be called here
        .send()
        .await
        .expect("request is successful");

    println!("{res:?}");
}

Configuring runtime plugins for generic clients

Generic clients can be configured in nearly all of the ways that SDK clients can. However, they also get access to a powerful tool that SDKs don’t have: Runtime plugins. Runtime plugins are similar to interceptors, except that they’re focused on building configuration for the request/response pipeline, instead of running inside it.

Runtime plugins can configure:

• Runtime Components — sub-functions of the orchestrator, responsible for specific parts of the request/response lifecycle. This includes things like:
  • The retry classifiers and the retry strategy,
  • A connector used to send requests and manage connections,
  • Identity resolution, auth schemes, and the auth option resolver,
  • Endpoint resolution,
  • System dependencies like a time source and sleep implementation,
• Interceptors — Hook-run functions for modifying and augmenting the request/response lifecycle. These were discussed in a previous section.
• Configuration — Low-level access to configuration, with the ability to replace, modify, or disable any configuration (excepting configuration based on private structs and enums that was inserted by other runtime plugins.)

To better understand the power of runtime plugins, let’s look at a few examples.

Runtime plugins can bundle configuration and interceptors

To get started writing runtime plugins, you’ll need to create a struct (or enum) and have it implement the RuntimePlugin trait:

#[derive(Debug)]
struct ExampleRuntimePlugin {
    // Configuration is stored in a "frozen" config bag layer. The config bag stores
    // settings that can be changed until they're "frozen". Once frozen, they can't
    // be changed, but they can be overridden. Because of this rule, it's very easy
    // to understand how a config bag value was set: values will always be associated
    // with the code that set them. If a value is being overridden, 'debug printing'
    // it will reveal why that's the case.
    config: Option<config_bag::FrozenLayer>,
    // Runtime plugins can configure runtime components. Some components can have
    // multiple values at once (like auth schemes.) Other components are singletons,
    // and can only be replaced or unset (like the retry strategy.)
    runtime_components: RuntimeComponentsBuilder,
}

#[derive(Debug)]
struct ExampleConfigStruct; 

impl Storable for ExampleConfigStruct {
    type Storer = StoreReplace<Self>;
}

impl ExampleRuntimePlugin {
    pub fn new() -> Self {
        // It's best to create your config layer, interceptors, and runtime
        // components here.
        let mut runtime_components =
            RuntimeComponentsBuilder::new("ExampleRuntimePlugin");
        runtime_components.push_interceptor(
            SharedInterceptor::new(ExampleInterceptor::new()) as _
        );
        let mut config = Layer::new("ExampleRuntimePlugin");
        config.store_put(ExampleConfigStruct);
        
        Self {
            config: Some(config.freeze()),
            runtime_components,
        }
    }
}

impl RuntimePlugin for ExampleRuntimePlugin {
    // While it's OK to create a new frozen layer each time this is called, it's
    // more performant to create the frozen layer once, and copy it as needed.
    // Because it's frozen copying it doesn't carry a risk that configs will
    // change after the copy has occurred, which would cause difficult-to-diagnose
    // bugs.
    fn config(&self) -> Option<config_bag::FrozenLayer> {
        self.config.clone()
    }

    // Interceptors and runtime components both can be set here, but it's better to
    // create them once and return references to them. That way, the cost of creation
    // is only paid once.
    fn runtime_components(&self) -> Cow<'_, RuntimeComponentsBuilder> {
        Cow::Borrowed(&self.runtime_components)
    }
}

Runtime plugins can replace the retry strategy and classifiers

It’s possible to replace default runtime components with your own versions. First, we’ll define a new retry strategy and a retry classifier. The classifier isn’t technically necessary, because it’s up to the retry strategy to call on it. We define one here
because it’s a good practice to separate the classification logic from the logic that decides how to respond to the classification result.

struct ExampleRetryStrategy;
struct ExampleRetryClassifier;

impl RetryStrategy for ExampleRetryStrategy {
    fn should_attempt_initial_request(
        &self,
        _runtime_components: &RuntimeComponents,
        _cfg: &ConfigBag,
    ) -> Result<ShouldAttempt, BoxError> {
        Ok(ShouldAttempt::Yes)
    }
    
    // This is a basic retry example that does most things required
    // by a retry strategy.
    fn should_attempt_retry(
        &self,
        ctx: &InterceptorContext,
        runtime_components: &RuntimeComponents,
        cfg: &ConfigBag,
    ) -> Result<ShouldAttempt, BoxError> {
        // Look a the result. If it's OK then we're done; No retry required. Otherwise, we need to inspect it
        let output_or_error = ctx.output_or_error().expect(
            "This is never be called without reaching the point where the result exists.",
        );
        if output_or_error.is_ok() {
            tracing::trace!("request succeeded, no retry necessary");
            return Ok(ShouldAttempt::No);
        }

        // Check if we're out of attempts
        let request_attempts = cfg
            .load::<RequestAttempts>()
            .expect("at least one request attempt is made before any retry is attempted");
        if request_attempts.attempts() >= self.max_attempts {
            tracing::trace!(
                attempts = request_attempts.attempts(),
                max_attempts = self.max_attempts,
                "not retrying because we are out of attempts"
            );
            return Ok(ShouldAttempt::No);
        }

        let retry_classifiers = runtime_components
            .retry_classifiers()
            .expect("a retry classifier is set");
        let retry_reason = retry_classifiers.classify_retry(ctx);

        let backoff = /* Calculate your own delay based on the retry reason */;

        tracing::debug!(
            "attempt {} failed with {:?}; retrying after {:?}",
            request_attempts.attempts(),
            retry_reason,
            backoff
        );

        Ok(ShouldAttempt::YesAfterDelay(backoff))
    }
}

impl ClassifyRetry for ExampleRetryClassifier {
    fn classify_retry(&self, ctx: &InterceptorContext) -> Option<RetryReason> {
        // examine the interceptor context and determine if a retry is necessary.
        // If yes, return `Some(RetryReason)`. Otherwise, return `None`.
    }

    // Because multiple classifiers may be composed together, each one is given a
    // name to aid in debugging.
    fn name(&self) -> &'static str {
        "Example Retry Classifier"
    }
}

Now that we have a classifier and strategy defined, we create a runtime plugin that will set both of them:

#[derive(Debug)]
struct CustomRetriesRuntimePlugin {
    runtime_components: RuntimeComponentsBuilder,
}

impl CustomRetriesRuntimePlugin {
    pub fn new() -> Self {
        // RetryClassifiers composes retry classifiers together. Each classifier
        // will be run in turn. As soon as one returns a retry reason, execution
        // short circuits and ends. If none return a reason, then the request
        // shouldn't be retried. Note that the retry strategy could be written such
        // that it ignores the result of classification or the classifier entirely.
        let retry_classifiers = RetryClassifiers::new()
            .with_classifier(ExampleRetryClassifier);
        
        let mut runtime_components =
            RuntimeComponentsBuilder::new("CustomRetriesRuntimePlugin")
                .with_retry_classifiers(retry_classifiers)
                .with_retry_strategy(ExampleRetryStrategy);

        Self { runtime_components }
    }
}

impl RuntimePlugin for CustomRetriesRuntimePlugin {
    // fn config() has a default impl that does nothing,
    // so we don't need to implement it.

    fn runtime_components(&self) -> Cow<'_, RuntimeComponentsBuilder> {
        Cow::Borrowed(&self.runtime_components)
    }
}

Runtime plugins can insert shared state into the config bag

Runtime plugins are also useful for managing state that must be shared across different requests. The built-in SDK retry strategy is a great example of this, as it utilizes shared state for an client rate limiter and token bucket. Because the rate limiter and token bucket state is shared across requests, throttling responses received in one request/response lifecycle can affect how responses are retried in a different request/response lifecycle. This is only desirable when both requests are to the same service, so this state is “partitioned” based on the service’s name. In this way, request state is only shared on a per-service basis.

If you need to share state related to your client’s environmental constraints, then partitioning shared state may not be desirable at all. However, in both cases, sharing state is accomplished by passing around clones of a singleton wrapped in a std::sync::Arc and a std::sync::Mutex:

#[derive(Debug)]
struct SharedStateRuntimePlugin {
    shared_state: SharedState
};

#[derive(Debug, Clone)]
struct SharedState {
    inner: Arc<Mutex<Inner>>
}

#[derive(Debug)]
struct Inner {
    // Define your shared state here
}

impl Storable for SharedState {
    type Storer = StoreReplace<Self>;
}

impl SharedStateRuntimePlugin {
        let shared_state = SharedState {
            inner: Arc::new(Mutex::new(Inner {
                // Init your shared state here
            }))
        };

        Self { shared_state }
    }
}

impl RuntimePlugin for SharedStateRuntimePlugin {
    fn config(&self) -> Option<config_bag::FrozenLayer> {
        let mut config = Layer::new("SharedStateRuntimePlugin");
        config.store_put(self.shared_state.clone());

        Some(config)
    }
    
    // fn runtime_components() has a default impl that does nothing,
    // so we don't need to implement it.
}

Once the shared state is in the config bag (and is a public type), any interceptor can access it.

CustomizableOperation::map_operation has been removed

A customizable operation allows a user to modify the current operation before it is dispatched. map_request and mutate_request on CustomizableOperation can be used in the orchestrator in the exactly the same way as before. However, map_operation will no longer be supported since the orchestrator does not use the aws_smithy_http::operation::Operation struct.

The following examples demonstrate how to upgrade existing code that uses map_operation in order to achieve the same functionality in the orchestrator.

Example #1: Inserting an item into a property bag

Previously map_operation allowed users to put an item into aws_smithy_http::property_bag::PropertyBag like so:

let sdk_config: aws_types::sdk_config::SdkConfig = /* create an SdkConfig */;
let s3_client = aws_sdk_s3::client::Client::new(&sdk_config);
s3_client
    .get_object()
    .customize()
    .await?
    .map_operation(|mut op| {
        op.properties_mut()
            .insert(SigningService::from_static(/* a string for signing */));
        Ok::<_, Infallible>(op)
    })
    .unwrap()
    .send()
    .await;

This will no longer compile in the orchestrator because Operation will cease to exist. You can upgrade the above code as follows (or any use case that inserts an item into the PropertyBag for that matter):

// Somewhere in your crate
use aws_smithy_runtime_api::box_error::BoxError;
use aws_smithy_runtime_api::client::interceptors::Interceptor;
use aws_smithy_runtime_api::client::interceptors::context::BeforeTransmitInterceptorContextMut;
use aws_smithy_types::config_bag::ConfigBag;

#[derive(Debug)]
struct CustomSigningInterceptor;

impl Interceptor for CustomSigningInterceptor {
    fn name(&self) -> &'static str {
        "CustomSigningInterceptor"
    }

     fn modify_before_signing(
        &self,
        // The context allows access to inputs, requests, responses, and outputs
        // depending on the hook. We don't need to access those in this example.
        _context: &mut BeforeTransmitInterceptorContextMut<ˈ_>,
        _runtime_components: &RuntimeComponents,
        cfg: &mut ConfigBag,
    ) -> Result<(), BoxError> {
        cfg
            .interceptor_state()
            .store_put(SigningService::from_static(/* a string for signing */));
    }
}

// Use `CustomSigningInterceptor` as follows

let sdk_config: aws_types::sdk_config::SdkConfig = /* create an SdkConfig */;
let s3_client = aws_sdk_s3::client::Client::new(&sdk_config);
s3_client
    .get_object()
    .customize()
    .await?
    .interceptor(CustomSigningInterceptor)
    .send()
    .await;

Just remember that items to be stored using store_put need to implement the Storable trait (here is an example of how a type can be made Storable).
Furthermore, the Interceptor trait defines several methods, and which one to use depends on the existing code to upgrade (the example above happens to use modify_before_signing). Refer back to What hook(s) should I implement for my interceptor? for more details.

Example #2: Extracting information from aws_smithy_http::operation::Metadata

map_operation used to allow users to extract fields in a Metadata, which is a struct attached to an operation that identifies the API being called. It can be useful to obtain a service name and an operation name for a use case such as metrics collection. For instance,

let sdk_config: aws_types::sdk_config::SdkConfig = /* create an SdkConfig */;
let s3_client = aws_sdk_s3::client::Client::new(&sdk_config);
s3_client
    .get_object()
    .customize()
    .await?
    .map_operation(|operation| {
        operation.try_clone().map(|operation| {
            let (_, parts) = operation.into_request_response();
            parts.metadata.map(|metadata| {
                let service_name = metadata.service().to_string().to_uppercase();
                let operation_name = metadata.name().to_string();
                /*
                 * do something with `service_name` and `operation_name`
                 */
            })
        });

        Ok(operation)
    })?
    .send()
    .await;

This piece of code can be upgraded as follows. You need to define your interceptor that extracts a Metadata out of ConfigBag:

// Somewhere in your crate
use aws_smithy_runtime_api::box_error::BoxError;
use aws_smithy_runtime_api::client::interceptors::Interceptor;
use aws_smithy_runtime_api::client::interceptors::context::BeforeTransmitInterceptorContextMut;
use aws_smithy_types::config_bag::ConfigBag;

#[derive(Debug)]
struct ServiceOperationNamesInterceptor;

impl Interceptor for ServiceOperationNamesInterceptor {
    fn name(&self) -> &'static str {
        "ServiceOperationNamesInterceptor"
    }

    // We show `modify_before_signing` for the purpose of this example.
    // Choose a different trait method to implement depending on the use case.
    fn modify_before_signing(
        &self,
        _context: &mut BeforeTransmitInterceptorContextMut<ˈ_>,
        _runtime_components: &RuntimeComponents,
        cfg: &mut ConfigBag,
    ) -> Result<(), BoxError> {
        let metadata = cfg.load::<aws_smithy_http::operation::Metadata>().expect("metadata should exist");
        let service_name = metadata.service().to_string();
        let operation_name = metadata.name().to_string();
        /*
         * do something with `service_name` and `operation_name`
         */ 
    }
}

// Use `ServiceOperationNamesInterceptor` as follows

let sdk_config: aws_types::sdk_config::SdkConfig = /* create an SdkConfig */;
let s3_client = aws_sdk_s3::client::Client::new(&sdk_config);
s3_client
    .get_object()
    .customize()
    .await?
    .interceptor(ServiceOperationNamesInterceptor)
    .send()
    .await;

Custom Auth Schemes

Previously, custom auth schemes would have been implemented by providing a middleware stage that resolves an identity and signs the request. Now there are IdentityResolver, AuthScheme, and Signer traits (in aws-smithy-runtime-api) that need to be implemented and registered as runtime components during client configuration.

All the examples in this section will illustrate how to implement HTTP Basic Auth as a custom auth scheme.

Identity resolvers

Identity resolvers must implement the IdentityResolver trait, which has an async resolve_identity function that returns an Identity. Any struct can be stored in an Identity, so you should create a struct for it that holds the information needed to implement your signer. Below is an example Login identity that could be used with Basic Auth:

#[derive(Clone, Eq, PartialEq)]
struct Login {
    user: String,
    password: Zeroizing<String>,
}

impl Debug for Login {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Login")
            .field("user", &self.0.user)
            .field("password", &"** redacted **")
            .finish()
    }
}

Important: Manually implement Debug with redaction so that you don’t unintentionally log secrets in your application.

Where an identity resolves from varies from application to application. The below example illustrates resolving the above Login from environment variables:

#[derive(Debug)]
pub struct BasicAuthIdentityResolver;

impl IdentityResolver for BasicAuthIdentityResolver {
    fn resolve_identity(&self, config_bag: &ConfigBag) -> Future<Identity> {
        match (std::env::var("EXAMPLE_USER"), std::env::var("EXAMPLE_PASS")) {
            (Ok(user), Ok(pass)) => {
                Future::ready(Ok(Identity::new(
                    Login {
                        user,
                        password: Zeroizing::new(pass),
                    },
                    // This simple example doesn't illustrate identity expiration. But if you needed it,
                    // you would set it below:
                    None,
                )))
            }
            _ => Future::ready(Err("missing username or password for basic auth".into())),
        }
    }
}

Important: As of this release, identities are not cached. If you need caching, you will have to implement it yourself. We will be adding identity resolver caching in a future release.

Signers

A signer is given a resolved identity and must sign a request with it. For basic auth, this may look as follows:

#[derive(Debug, Default)]
struct BasicAuthSigner;

impl Signer for BasicAuthSigner {
    fn sign_http_request(
        &self,
        request: &mut HttpRequest,
        identity: &Identity,
        // In some advanced use cases, the Smithy `@endpointRuleSet` can
        // provide additional context in this config:
        _auth_scheme_endpoint_config: AuthSchemeEndpointConfig<'_>,
        _runtime_components: &RuntimeComponents,
        _config_bag: &ConfigBag,
    ) -> Result<(), BoxError> {
        // Downcast the identity to our known `Login` type
        let login = identity
            .data::<Login>()
            .ok_or("HTTP basic auth requires a `Login` identity")?;
        // Use the login to add an authorization header to the request
        request.headers_mut().insert(
            http::header::AUTHORIZATION,
            HeaderValue::from_str(&format!(
                "Basic {}",
                base64::encode(format!("{}:{}", login.user(), login.password()))
            ))
            .expect("valid header value"),
        );
        Ok(())
    }
}

Auth schemes

The auth scheme is responsible for telling the client which identity resolver and signer to use. This implementation will typically be very simple:

#[derive(Debug, Default)]
pub struct BasicAuthScheme {
    // We'll get to the signer later...
    signer: BasicAuthSigner,
}

impl AuthScheme for BasicAuthScheme {
    fn scheme_id(&self) -> AuthSchemeId {
        // Make sure this auth scheme ID is consistent throughout.
        // Consider making a constant for it.
        AuthSchemeId::new("basicauth")
    }

    fn identity_resolver(
        &self,
        identity_resolvers: &dyn GetIdentityResolver,
    ) -> Option<SharedIdentityResolver> {
        // Most auth scheme implementations will determine the identity resolver
        // as illustrated below. More advanced use cases may need to choose between
        // multiple identity resolvers.
        identity_resolvers.identity_resolver(self.scheme_id())
    }

    fn signer(&self) -> &dyn Signer {
        &self.signer
    }
}

Configuration

Piecing everything together requires implementing the RuntimePlugin trait, and registering this runtime plugin in the client config. Note that if you need different auth schemes per operation, and want them modeled in the Smithy model, then you will need to write a smithy-rs codegen plugin. Since the code generator’s plugin API isn’t even remotely stable right now, this won’t be covered here.

First, implementing the RuntimePlugin trait:

#[derive(Debug)]
pub struct BasicAuthRuntimePlugin {
    components: RuntimeComponentsBuilder,
}

impl BasicAuthRuntimePlugin {
    // It's generally recommended to create the runtime components in the runtime plugin's
    // constructor so that they aren't recreated for every request that's made.
    pub fn new() -> Self {
        let scheme_id = AuthSchemeId::new("basicauth");
        Self {
            components: RuntimeComponentsBuilder::new()
                // Register our auth scheme
                .with_auth_scheme(SharedAuthScheme::new(BasicAuthAuthScheme))
                // Register our identity resolver with our auth scheme ID
                .with_identity_resolver(
                    // This scheme ID needs to match the scheme ID returned in the auth scheme implementation
                    scheme_id,
                    SharedIdentityResolver::new(BasicAuthIdentityResolver),
                )
                // Set the auth scheme option resolver to always use our basic auth auth scheme
                .with_auth_scheme_option_resolver(Some(SharedAuthSchemeOptionResolver::new(
                    StaticAuthSchemeOptionResolver::new(vec![scheme_id]),
                ))),
        }
    }
}

impl RuntimePlugin for BasicAuthRuntimePlugin {
    // Runtime components are configurable implementations that the client
    // can use to fulfill a request. This `runtime_components` method allows
    // us to set the auth/identity related components.
    fn runtime_components(&self) -> Cow<'_, RuntimeComponentsBuilder> {
        Cow::Borrowed(&self.components)
    }
}

Finally, during client construction, this runtime plugin needs to be given to the client config:

let config = my_service::Config::builder()
    .runtime_plugin(BasicAuthRuntimePlugin::new())
    // ... set other required things such as the connector
    .build();
let client = my_service::Client::from_conf(config);

At that point, everything should be wired up for your custom auth scheme.

One last thing: those generating their own client with smithy-rs may temporarily opt-out of these changes

For a limited time, we’re making it possible for users to upgrade their smithy-rs version but to keep using the old middleware-based client. You can do this by opening your smithy-build.json and setting "enableNewSmithyRuntime" to "middleware" (as opposed to its default, "orchestrator"). The next time you generate a client, it’ll use the old middleware implementation. For example:

{
  "version": "1.0",
  "projections": {
    "config": {
      "plugins": {
        "rust-client-codegen": {
          "codegen": {
            "enableNewSmithyRuntime": "middleware",
            -- snip --
          },
          -- snip --
        }
      }
    }
  },
  -- snip --
}

[deven]

The documentation links pointing to awslabs.github.io appear to be restricted access only. The following links are open to the public:

• RFC-0034: Smithy Orchestrator
• What is the ‘orchestrator’ and why does it exist?
• Identity and Auth in Clients