[smithy-rs] feature: orchestrator retry classifiers (#2621)

## Motivation and Context
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here -->
To retry a response, we must first classify it as retryable.

## Description
<!--- Describe your changes in detail -->
feature: add AWS error code classifier
feature: add x-amz-retry-after header classifier
feature: add smithy modeled retry classifier
feature: add error type classifier
feature: add HTTP status code classifier
add: tests for classifiers
remove: redundant `http` dep from `aws-http`
move: `NeverRetryStrategy` to smithy-runtime crate
add: RuntimePluginImpls codegen section for operation-specific runtime
plugin definitions
update: orchestrator retries to work with `ShouldAttempt`
add: retry classifier config bag accessor
add: raw response getter to SdkError
update: RetryStrategy trait signatures to use `ShouldAttempt`
add: `RetryClassifiers` struct for holding and calling retry classifiers
update: `RetryClassifierDecorator` to define orchestrator classifiers
add: `default_retry_classifiers` fn to codegen
update: `ServiceGenerator` to add feature flag for
aws-smithy-runtime/test-util
update: SRA integration test to insert retry classifier plugin

## Testing
<!--- Please describe in detail how you tested your changes -->
<!--- Include details of your testing environment, and the tests you ran
to -->
<!--- see how your change affects other areas of the code, etc. -->
this change includes tests

----

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

Author: Zelda Hessler

sdk/aws-http/Cargo.toml

@@ -41,7 +41,6 @@ version = "0.55.2"
 async-trait = "0.1.50"
 bytes-utils = "0.1.2"
 env_logger = "0.9"
-http = "0.2.3"
 proptest = "1"
 serde_json = "1"
 

sdk/aws-runtime/external-types.toml

@@ -1,7 +1,8 @@
 allowed_external_types = [
     "aws_credential_types::*",
     "aws_sigv4::*",
-    "aws_smithy_http::body::SdkBody",
+    "aws_smithy_http::*",
+    "aws_smithy_types::*",
     "aws_smithy_runtime_api::*",
     "aws_types::*",
     "http::request::Request",

sdk/aws-runtime/src/lib.rs

@@ -25,5 +25,8 @@ pub mod recursion_detection;
 /// Supporting code for user agent headers in the AWS SDK.
 pub mod user_agent;
 
+/// Supporting code for retry behavior specific to the AWS SDK.
+pub mod retries;
+
 /// Supporting code for invocation ID headers in the AWS SDK.
 pub mod invocation_id;

sdk/aws-runtime/src/retries.rs

@@ -0,0 +1,7 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/// Classifiers that can inspect a response and determine if it should be retried.
+pub mod classifier;

sdk/aws-runtime/src/retries/classifier.rs

@@ -0,0 +1,174 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+use aws_smithy_http::http::HttpHeaders;
+use aws_smithy_http::result::SdkError;
+use aws_smithy_runtime_api::client::retries::RetryReason;
+use aws_smithy_types::error::metadata::ProvideErrorMetadata;
+use aws_smithy_types::retry::ErrorKind;
+
+/// AWS error codes that represent throttling errors.
+pub const THROTTLING_ERRORS: &[&str] = &[
+    "Throttling",
+    "ThrottlingException",
+    "ThrottledException",
+    "RequestThrottledException",
+    "TooManyRequestsException",
+    "ProvisionedThroughputExceededException",
+    "TransactionInProgressException",
+    "RequestLimitExceeded",
+    "BandwidthLimitExceeded",
+    "LimitExceededException",
+    "RequestThrottled",
+    "SlowDown",
+    "PriorRequestNotComplete",
+    "EC2ThrottledException",
+];
+
+/// AWS error codes that represent transient errors.
+pub const TRANSIENT_ERRORS: &[&str] = &["RequestTimeout", "RequestTimeoutException"];
+
+/// A retry classifier for determining if the response sent by an AWS service requires a retry.
+#[derive(Debug)]
+pub struct AwsErrorCodeClassifier;
+
+impl AwsErrorCodeClassifier {
+    /// Classify an error code to check if represents a retryable error. The codes of retryable
+    /// errors are defined [here](THROTTLING_ERRORS) and [here](TRANSIENT_ERRORS).
+    pub fn classify_error<E: ProvideErrorMetadata, R>(
+        &self,
+        error: &SdkError<E, R>,
+    ) -> Option<RetryReason> {
+        if let Some(error_code) = error.code() {
+            if THROTTLING_ERRORS.contains(&error_code) {
+                return Some(RetryReason::Error(ErrorKind::ThrottlingError));
+            } else if TRANSIENT_ERRORS.contains(&error_code) {
+                return Some(RetryReason::Error(ErrorKind::TransientError));
+            }
+        };
+
+        None
+    }
+}
+
+/// A retry classifier that checks for `x-amz-retry-after` headers. If one is found, a
+/// [`RetryReason::Explicit`] is returned containing the duration to wait before retrying.
+#[derive(Debug)]
+pub struct AmzRetryAfterHeaderClassifier;
+
+impl AmzRetryAfterHeaderClassifier {
+    /// Classify an AWS responses error code to determine how (and if) it should be retried.
+    pub fn classify_error<E>(&self, error: &SdkError<E>) -> Option<RetryReason> {
+        error
+            .raw_response()
+            .and_then(|res| res.http_headers().get("x-amz-retry-after"))
+            .and_then(|header| header.to_str().ok())
+            .and_then(|header| header.parse::<u64>().ok())
+            .map(|retry_after_delay| {
+                RetryReason::Explicit(std::time::Duration::from_millis(retry_after_delay))
+            })
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::{AmzRetryAfterHeaderClassifier, AwsErrorCodeClassifier};
+    use aws_smithy_http::body::SdkBody;
+    use aws_smithy_http::operation;
+    use aws_smithy_http::result::SdkError;
+    use aws_smithy_runtime_api::client::retries::RetryReason;
+    use aws_smithy_types::error::metadata::ProvideErrorMetadata;
+    use aws_smithy_types::error::ErrorMetadata;
+    use aws_smithy_types::retry::{ErrorKind, ProvideErrorKind};
+    use std::fmt;
+    use std::time::Duration;
+
+    #[derive(Debug)]
+    struct UnmodeledError;
+
+    impl fmt::Display for UnmodeledError {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+            write!(f, "UnmodeledError")
+        }
+    }
+
+    impl std::error::Error for UnmodeledError {}
+
+    struct CodedError {
+        metadata: ErrorMetadata,
+    }
+
+    impl CodedError {
+        fn new(code: &'static str) -> Self {
+            Self {
+                metadata: ErrorMetadata::builder().code(code).build(),
+            }
+        }
+    }
+
+    impl ProvideErrorKind for UnmodeledError {
+        fn retryable_error_kind(&self) -> Option<ErrorKind> {
+            None
+        }
+
+        fn code(&self) -> Option<&str> {
+            None
+        }
+    }
+
+    impl ProvideErrorMetadata for CodedError {
+        fn meta(&self) -> &ErrorMetadata {
+            &self.metadata
+        }
+    }
+
+    #[test]
+    fn classify_by_error_code() {
+        let policy = AwsErrorCodeClassifier;
+        let res = http::Response::new("OK");
+        let err = SdkError::service_error(CodedError::new("Throttling"), res);
+
+        assert_eq!(
+            policy.classify_error(&err),
+            Some(RetryReason::Error(ErrorKind::ThrottlingError))
+        );
+
+        let res = http::Response::new("OK");
+        let err = SdkError::service_error(CodedError::new("RequestTimeout"), res);
+        assert_eq!(
+            policy.classify_error(&err),
+            Some(RetryReason::Error(ErrorKind::TransientError))
+        )
+    }
+
+    #[test]
+    fn classify_generic() {
+        let policy = AwsErrorCodeClassifier;
+        let res = http::Response::new("OK");
+        let err = aws_smithy_types::Error::builder().code("SlowDown").build();
+        let err = SdkError::service_error(err, res);
+        assert_eq!(
+            policy.classify_error(&err),
+            Some(RetryReason::Error(ErrorKind::ThrottlingError))
+        );
+    }
+
+    #[test]
+    fn test_retry_after_header() {
+        let policy = AmzRetryAfterHeaderClassifier;
+        let res = http::Response::builder()
+            .header("x-amz-retry-after", "5000")
+            .body("retry later")
+            .unwrap()
+            .map(SdkBody::from);
+        let res = operation::Response::new(res);
+        let err = SdkError::service_error(UnmodeledError, res);
+
+        assert_eq!(
+            policy.classify_error(&err),
+            Some(RetryReason::Explicit(Duration::from_millis(5000))),
+        );
+    }
+}

sdk/aws-smithy-http/src/result.rs

@@ -5,15 +5,17 @@
 
 //! `Result` wrapper types for [success](SdkSuccess) and [failure](SdkError) responses.
 
-use crate::connection::ConnectionMetadata;
-use crate::operation;
-use aws_smithy_types::error::metadata::{ProvideErrorMetadata, EMPTY_ERROR_METADATA};
-use aws_smithy_types::error::ErrorMetadata;
-use aws_smithy_types::retry::ErrorKind;
 use std::error::Error;
 use std::fmt;
 use std::fmt::{Debug, Display, Formatter};
 
+use aws_smithy_types::error::metadata::{ProvideErrorMetadata, EMPTY_ERROR_METADATA};
+use aws_smithy_types::error::ErrorMetadata;
+use aws_smithy_types::retry::ErrorKind;
+
+use crate::connection::ConnectionMetadata;
+use crate::operation;
+
 type BoxError = Box<dyn Error + Send + Sync>;
 
 /// Successful SDK Result
@@ -434,6 +436,15 @@ impl<E, R> SdkError<E, R> {
         }
     }
 
+    /// Return a reference to this error's raw response, if it contains one. Otherwise, return `None`.
+    pub fn raw_response(&self) -> Option<&R> {
+        match self {
+            Self::ServiceError(inner) => Some(inner.raw()),
+            Self::ResponseError(inner) => Some(inner.raw()),
+            _ => None,
+        }
+    }
+
     /// Maps the service error type in `SdkError::ServiceError`
     #[doc(hidden)]
     pub fn map_service_error<E2>(self, map: impl FnOnce(E) -> E2) -> SdkError<E2, R> {

sdk/aws-smithy-runtime-api/src/client.rs

@@ -18,6 +18,7 @@ pub mod orchestrator;
 /// This code defines when and how failed requests should be retried. It also defines the behavior
 /// used to limit the rate that requests are sent.
 pub mod retries;
+
 /// Runtime plugin type definitions.
 pub mod runtime_plugin;
 

sdk/aws-smithy-runtime-api/src/client/orchestrator.rs

@@ -6,7 +6,8 @@
 use super::identity::{IdentityResolver, IdentityResolvers};
 use crate::client::identity::Identity;
 use crate::client::interceptors::context::{Input, OutputOrError};
-use crate::client::interceptors::InterceptorContext;
+use crate::client::retries::RetryClassifiers;
+use crate::client::retries::RetryStrategy;
 use crate::config_bag::ConfigBag;
 use crate::type_erasure::{TypeErasedBox, TypedBox};
 use aws_smithy_async::future::now_or_later::NowOrLater;
@@ -54,16 +55,6 @@ impl Connection for Box<dyn Connection> {
     }
 }
 
-pub trait RetryStrategy: Send + Sync + Debug {
-    fn should_attempt_initial_request(&self, cfg: &ConfigBag) -> Result<(), BoxError>;
-
-    fn should_attempt_retry(
-        &self,
-        context: &InterceptorContext<HttpRequest, HttpResponse>,
-        cfg: &ConfigBag,
-    ) -> Result<bool, BoxError>;
-}
-
 #[derive(Debug)]
 pub struct AuthOptionResolverParams(TypeErasedBox);
 
@@ -241,6 +232,9 @@ pub trait ConfigBagAccessors {
         response_serializer: impl ResponseDeserializer + 'static,
     );
 
+    fn retry_classifiers(&self) -> &RetryClassifiers;
+    fn set_retry_classifiers(&mut self, retry_classifier: RetryClassifiers);
+
     fn retry_strategy(&self) -> &dyn RetryStrategy;
     fn set_retry_strategy(&mut self, retry_strategy: impl RetryStrategy + 'static);
 
@@ -277,25 +271,6 @@ impl ConfigBagAccessors for ConfigBag {
         self.put::<Box<dyn AuthOptionResolver>>(Box::new(auth_option_resolver));
     }
 
-    fn http_auth_schemes(&self) -> &HttpAuthSchemes {
-        self.get::<HttpAuthSchemes>()
-            .expect("auth schemes must be set")
-    }
-
-    fn set_http_auth_schemes(&mut self, http_auth_schemes: HttpAuthSchemes) {
-        self.put::<HttpAuthSchemes>(http_auth_schemes);
-    }
-
-    fn retry_strategy(&self) -> &dyn RetryStrategy {
-        &**self
-            .get::<Box<dyn RetryStrategy>>()
-            .expect("a retry strategy must be set")
-    }
-
-    fn set_retry_strategy(&mut self, retry_strategy: impl RetryStrategy + 'static) {
-        self.put::<Box<dyn RetryStrategy>>(Box::new(retry_strategy));
-    }
-
     fn endpoint_resolver_params(&self) -> &EndpointResolverParams {
         self.get::<EndpointResolverParams>()
             .expect("endpoint resolver params must be set")
@@ -334,6 +309,15 @@ impl ConfigBagAccessors for ConfigBag {
         self.put::<Box<dyn Connection>>(Box::new(connection));
     }
 
+    fn http_auth_schemes(&self) -> &HttpAuthSchemes {
+        self.get::<HttpAuthSchemes>()
+            .expect("auth schemes must be set")
+    }
+
+    fn set_http_auth_schemes(&mut self, http_auth_schemes: HttpAuthSchemes) {
+        self.put::<HttpAuthSchemes>(http_auth_schemes);
+    }
+
     fn request_serializer(&self) -> &dyn RequestSerializer {
         &**self
             .get::<Box<dyn RequestSerializer>>()
@@ -357,6 +341,25 @@ impl ConfigBagAccessors for ConfigBag {
         self.put::<Box<dyn ResponseDeserializer>>(Box::new(response_deserializer));
     }
 
+    fn retry_classifiers(&self) -> &RetryClassifiers {
+        self.get::<RetryClassifiers>()
+            .expect("retry classifiers must be set")
+    }
+
+    fn set_retry_classifiers(&mut self, retry_classifiers: RetryClassifiers) {
+        self.put::<RetryClassifiers>(retry_classifiers);
+    }
+
+    fn retry_strategy(&self) -> &dyn RetryStrategy {
+        &**self
+            .get::<Box<dyn RetryStrategy>>()
+            .expect("a retry strategy must be set")
+    }
+
+    fn set_retry_strategy(&mut self, retry_strategy: impl RetryStrategy + 'static) {
+        self.put::<Box<dyn RetryStrategy>>(Box::new(retry_strategy));
+    }
+
     fn trace_probe(&self) -> &dyn TraceProbe {
         &**self
             .get::<Box<dyn TraceProbe>>()