Retry additional classes of H2 errors (#2971)

Draft pull request pending merge of #2970 
## Motivation and Context
- awslabs/aws-sdk-rust#738
- awslabs/aws-sdk-rust#858
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here -->

## Description
<!--- Describe your changes in detail -->
This PR adds two additional classes of retries tested:
1. GO_AWAY: awslabs/aws-sdk-rust#738
2. REFUSED_STREAM: awslabs/aws-sdk-rust#858

I tested 1 by using the example helpfully provided. The fix for 2 is
untested and difficult to reproduce but since my fix for 1 worked, I'm
confident that we're detecting the correct error class here.

## Testing
I used the example provided by the customer and validated the H2 GO_AWAY
fix.

## Checklist
<!--- If a checkbox below is not applicable, then please DELETE it
rather than leaving it unchecked -->
- [ ] I have updated `CHANGELOG.next.toml` if I made changes to the
smithy-rs codegen or runtime crates
- [ ] 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._

rust-runtime/aws-smithy-client/Cargo.toml

@@ -14,7 +14,7 @@ wiremock = ["test-util", "dep:hyper", "hyper?/server", "hyper?/h2", "rustls", "t
 native-tls = []
 allow-compilation = [] # our tests use `cargo test --all-features` and native-tls breaks CI
 rustls = ["dep:hyper-rustls", "dep:lazy_static", "dep:rustls", "client-hyper", "rt-tokio"]
-client-hyper = ["dep:hyper", "hyper?/client", "hyper?/http2", "hyper?/http1", "hyper?/tcp"]
+client-hyper = ["dep:hyper", "hyper?/client", "hyper?/http2", "hyper?/http1", "hyper?/tcp", "dep:h2"]
 hyper-webpki-doctest-only = ["dep:hyper-rustls", "hyper-rustls?/webpki-roots"]
 
 [dependencies]
@@ -28,6 +28,7 @@ fastrand = "2.0.0"
 http = "0.2.3"
 http-body = "0.4.4"
 hyper = { version = "0.14.26", default-features = false, optional = true }
+h2 = { version = "0.3", default-features = false, optional = true }
 # cargo does not support optional test dependencies, so to completely disable rustls
 # we need to add the webpki-roots feature here.
 # https://github.com/rust-lang/cargo/issues/1596

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

@@ -92,6 +92,7 @@ use hyper::client::connect::{
     capture_connection, CaptureConnection, Connected, Connection, HttpInfo,
 };
 
+use h2::Reason;
 use std::error::Error;
 use std::fmt::Debug;
 
@@ -196,20 +197,28 @@ fn downcast_error(err: BoxError) -> ConnectorError {
 /// Convert a [`hyper::Error`] into a [`ConnectorError`]
 fn to_connector_error(err: hyper::Error) -> ConnectorError {
     if err.is_timeout() || find_source::<HttpTimeoutError>(&err).is_some() {
-        ConnectorError::timeout(err.into())
-    } else if err.is_user() {
-        ConnectorError::user(err.into())
-    } else if err.is_closed() || err.is_canceled() || find_source::<std::io::Error>(&err).is_some()
-    {
-        ConnectorError::io(err.into())
+        return ConnectorError::timeout(err.into());
+    }
+    if err.is_user() {
+        return ConnectorError::user(err.into());
+    }
+    if err.is_closed() || err.is_canceled() || find_source::<std::io::Error>(&err).is_some() {
+        return ConnectorError::io(err.into());
     }
     // We sometimes receive this from S3: hyper::Error(IncompleteMessage)
-    else if err.is_incomplete_message() {
-        ConnectorError::other(err.into(), Some(ErrorKind::TransientError))
-    } else {
-        tracing::warn!(err = %DisplayErrorContext(&err), "unrecognized error from Hyper. If this error should be retried, please file an issue.");
-        ConnectorError::other(err.into(), None)
+    if err.is_incomplete_message() {
+        return ConnectorError::other(err.into(), Some(ErrorKind::TransientError));
+    }
+    if let Some(h2_err) = find_source::<h2::Error>(&err) {
+        if h2_err.is_go_away()
+            || (h2_err.is_reset() && h2_err.reason() == Some(Reason::REFUSED_STREAM))
+        {
+            return ConnectorError::io(err.into());
+        }
     }
+
+    tracing::warn!(err = %DisplayErrorContext(&err), "unrecognized error from Hyper. If this error should be retried, please file an issue.");
+    ConnectorError::other(err.into(), None)
 }
 
 fn find_source<'a, E: Error + 'static>(err: &'a (dyn Error + 'static)) -> Option<&'a E> {