AWS SDK

rev. 1ed3d5371c342b375a8e2197cd2254b0dc08a51c (ignoring whitespace)

Files changed:

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/client/pool/registry.rs

@@ -797,797 +856,860 @@

             ConnectionReuseScope::Pool,
             NonZeroUsize::new(1),
-799
+        )
         .unwrap();
         let partition_1 = explicit_partition(&registry, 1);
         let partition_2 = explicit_partition(&registry, 2);
         let uri: Uri = "https://example.com/".parse().unwrap();
         let other: Uri = "https://other.example.com/".parse().unwrap();
-805
         let first = registry.resolve_cell(&partition_1, &uri).unwrap();
         let second = registry.resolve_cell(&partition_2, &uri).unwrap();
         let other = registry.resolve_cell(&partition_1, &other).unwrap();
-809
         let admission = first.admission().unwrap();
         assert!(Arc::ptr_eq(admission, second.admission().unwrap()));
         assert!(!Arc::ptr_eq(admission, other.admission().unwrap()));
-813
+    }
-814
     #[test]
     fn unbounded_origins_construct_no_admission_state() {

         let registry = explicit_registry([partition(1)], ConnectionReuseScope::Pool, None).unwrap();
         let partition = explicit_partition(&registry, 1);
         let cell = registry
             .resolve_cell(&partition, &"https://example.com/".parse().unwrap())
             .unwrap();
-822
         assert!(cell.admission().is_none());
         assert!(registry.admissions.lock().is_empty());
-825
+    }
-826
     /// Resolves the same first-use origin concurrently through one partition.
     ///
     /// Publication must retain one production `OriginCell` instance and return
     /// that exact instance to both callers.
     #[test]
     fn first_cell_publication_is_stable_under_contention() {
         const THREADS: usize = 8;
         let registry = Arc::new(
             explicit_registry(
                 [partition(1)],
                 ConnectionReuseScope::default(),
                 NonZeroUsize::new(1),
-839
+            )
             .unwrap(),

↧ Expand context

         );
         let partition = explicit_partition(&registry, 1);
         let barrier = Arc::new(Barrier::new(THREADS));
         let mut threads = Vec::new();
-845
         for _ in 0..THREADS {
             let registry = registry.clone();
             let partition = partition.clone();
             let barrier = barrier.clone();
             threads.push(std::thread::spawn(move || {
                 let uri: Uri = "https://example.com/".parse().unwrap();
                 barrier.wait();
                 registry.resolve_cell(&partition, &uri).unwrap()
             }));
-855
+        }
-856
         let cells: Vec<_> = threads
             .into_iter()
             .map(|thread| thread.join().unwrap())
             .collect();

@@ -870,874 +934,944 @@

↥ Expand context

 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use std::future::Future;
     use std::pin::Pin;
-879
     #[derive(Debug)]
     struct TestSpawner;
-882
     impl DriverSpawner for TestSpawner {
         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-886
+        }
-887
+    }
-888
     fn explicit_registry(
         partitions: impl IntoIterator<Item = Partition>,
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
     ) -> Result<PartitionRegistry, PartitionRegistryError> {

         let admission_policy =
             max_connections_per_host.map(|limit| AdmissionPolicy::new(limit, true));
         PartitionRegistry::new(
             Some(partitions.into_iter().collect()),
             reuse_scope,
             admission_policy,
             MaintenanceConfig::default(),
-901
+        )
-902
+    }
-903
     /// Resolves the same first-use origin through two concurrent production calls.
     ///
     /// Loom enumerates publication schedules while `PartitionRegistry` and
     /// `PartitionState` perform their ordinary lookup and insertion transitions.
     #[test]
     fn first_cell_publication_is_stable_under_contention() {
         loom::model(|| {
             let registry = Arc::new(
                 explicit_registry(
                     [Partition::new(PartitionId::from_index(1), TestSpawner)],
                     ConnectionReuseScope::default(),
                     None,
-916
+                )
                 .unwrap(),
             );
             let partition = registry.partition(PartitionId::from_index(1)).unwrap();
-920
             let first_registry = registry.clone();
             let first_partition = partition.clone();
             let first = loom::thread::spawn(move || {
                 let uri: Uri = "https://example.com/".parse().unwrap();
                 first_registry.resolve_cell(&first_partition, &uri).unwrap()
             });
             let second_registry = registry.clone();
             let second_partition = partition.clone();
             let second = loom::thread::spawn(move || {
                 let uri: Uri = "https://example.com/".parse().unwrap();
                 second_registry
                     .resolve_cell(&second_partition, &uri)
                     .unwrap()
             });
-935
             let first = first.join().unwrap();
             let second = second.join().unwrap();
             assert!(Arc::ptr_eq(&first, &second));
             assert_eq!(1, partition.cell_count());
             drop((first, second));
             partition.maintenance.clear_modeled_cells_for_test();
         });
-943
+    }
-944
+}

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/hyper_legacy.rs

@@ -338,338 +404,401 @@

↥ Expand context

-338
+    }
-339
+}
-340
 impl<C> HttpConnector for Adapter<C>
 where
     C: Clone + Send + Sync + 'static,
     C: hyper_0_14::service::Service<http_02x::Uri>,
     C::Response: Connection + AsyncRead + AsyncWrite + Send + Unpin + 'static,
     C::Future: Unpin + Send + 'static,
     C::Error: Into<BoxError>,
-348
+{
     fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
         use hyper_0_14::service::Service;
-351
         let mut request = match request.try_into_http02x() {
             Ok(request) => request,
             Err(err) => {
                 return HttpConnectorFuture::ready(Err(ConnectorError::other(err.into(), None)));
-356
+            }
         };

         let capture_connection = capture_connection(&mut request);
         if let Some(capture_smithy_connection) =
             request.extensions().get::<CaptureSmithyConnection>()
-361
+        {
             capture_smithy_connection
                 .set_connection_retriever(move || extract_smithy_connection(&capture_connection));
-364
+        }
         let mut client = self.client.clone();
         let fut = client.call(request);
         HttpConnectorFuture::new(async move {
             let response = fut
                 .await
                 .map_err(downcast_error)?
                 .map(SdkBody::from_body_0_4);
             match HttpResponse::try_from(response) {
                 Ok(response) => Ok(response),
                 Err(err) => Err(ConnectorError::other(err.into(), None)),
             match fut.await {
                 Ok(response) => HttpResponse::try_from(response.map(SdkBody::from_body_0_4))
                     .map_err(|err| ConnectorError::other(err.into(), None)),
                 Err(err) => Err(downcast_error(err)),
-372
+            }
         })
-374
+    }
-375
+}
-376
 /// Downcast errors coming out of hyper into an appropriate `ConnectorError`
 fn downcast_error(err: BoxError) -> ConnectorError {
     // is a `TimedOutError` (from aws_smithy_async::timeout) in the chain? if it is, this is a timeout
     if find_source::<TimedOutError>(err.as_ref()).is_some() {
         return ConnectorError::timeout(err);

↧ Expand context

-382
+    }
     // is the top of chain error actually already a `ConnectorError`? return that directly
     let err = match err.downcast::<ConnectorError>() {
         Ok(connector_error) => return *connector_error,
         Err(box_error) => box_error,
     };
     // generally, the top of chain will probably be a hyper error. Go through a set of hyper specific
     // error classifications
     let err = match err.downcast::<hyper_0_14::Error>() {
         Ok(hyper_error) => return to_connector_error(*hyper_error),
         Err(box_error) => box_error,
     };
-394
     // otherwise, we have no idea!
     ConnectorError::other(err, None)
-397
+}
-398
 /// Convert a [`hyper_0_14::Error`] into a [`ConnectorError`]
 fn to_connector_error(err: hyper_0_14::Error) -> ConnectorError {
     if err.is_timeout() || find_source::<HttpTimeoutError>(&err).is_some() {

@@ -464,461 +524,520 @@

↥ Expand context

             .field("connector_cache", &self.connector_cache)
             .field("client_builder", &self.client_builder)
             .finish()
-464
+    }
-465
+}
-466
 impl<C, F> HttpClient for HyperClient<F>
 where
     F: Fn() -> C + Send + Sync,
     C: Clone + Send + Sync + 'static,
     C: hyper_0_14::service::Service<http_02x::Uri>,
     C::Response: Connection + AsyncRead + AsyncWrite + Send + Unpin + 'static,
     C::Future: Unpin + Send + 'static,
     C::Error: Into<BoxError>,
-475
+{
     fn http_connector(
         &self,
         settings: &HttpConnectorSettings,
         components: &RuntimeComponents,
     ) -> SharedHttpConnector {

         let key = CacheKey::from(settings);
         let mut connector = self.connector_cache.read().unwrap().get(&key).cloned();
         if connector.is_none() {
             let mut cache = self.connector_cache.write().unwrap();
             // Short-circuit if another thread already wrote a connector to the cache for this key
             if !cache.contains_key(&key) {
                 let mut builder = HyperConnector::builder()
                     .hyper_builder(self.client_builder.clone())
                     .connector_settings(settings.clone());
                 builder.set_sleep_impl(components.sleep_impl());
--
                 let start = components.time_source().map(|ts| ts.now());
                 let tcp_connector = (self.tcp_connector_fn)();
                 let end = components.time_source().map(|ts| ts.now());
                 if let (Some(start), Some(end)) = (start, end) {
                     if let Ok(elapsed) = end.duration_since(start) {
                         tracing::debug!("new TCP connector created in {:?}", elapsed);
-497
+                    }
-498
+                }
                 let connector = SharedHttpConnector::new(builder.build(tcp_connector));
                 cache.insert(key.clone(), connector);

↧ Expand context

-501
+            }
             connector = cache.get(&key).cloned();
-503
+        }
-504
         connector.expect("cache populated above")
-506
+    }
-507
     fn validate_base_client_config(
         &self,
         _: &RuntimeComponentsBuilder,
         _: &ConfigBag,
     ) -> Result<(), BoxError> {
         // Initialize the TCP connector at this point so that native certs load
         // at client initialization time instead of upon first request. We do it
         // here rather than at construction so that it won't run if this is not
         // the selected HTTP client for the base config (for example, if this was
         // the default HTTP client, and it was overridden by a later plugin).
         let _ = (self.tcp_connector_fn)();
         Ok(())
-520
+    }

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/sync/loom.rs

@@ -1,1 +75,76 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Loom-backed synchronization primitives used by focused model tests.
 //! Loom-backed synchronization primitives used to run production pool state
 //! machines under enumerated thread schedules.
-8
 use ::std::ops::{Deref, DerefMut};
-10
 pub(crate) use loom::sync::atomic::{AtomicBool, Ordering};
 pub(crate) use loom::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 pub(crate) use loom::sync::Arc;
-13
 loom::thread_local! {
     static HELD_POOL_LOCKS: ::std::cell::Cell<usize> = ::std::cell::Cell::new(0);
-16
+}
-17
 struct LockDepth;
-19
 impl LockDepth {
     fn enter() -> Self {
         HELD_POOL_LOCKS.with(|depth| {
             assert_eq!(
 ,
                 depth.get(),
                 "pool coordination locks must never be nested"
             );
             depth.set(1);
         });
         Self
-31
+    }
-32
+}
-33
 impl Drop for LockDepth {
     fn drop(&mut self) {
         HELD_POOL_LOCKS.with(|depth| {
             assert_eq!(1, depth.get(), "pool lock-depth tracking became unbalanced");
             depth.set(0);
         });
-40
+    }
-41
+}
-42
 /// Loom substitute for a non-owning registry reference.
 ///
 /// Loom's modeled `Arc` has no weak-reference API. Keeping a strong reference
 /// in model tests preserves lookup behavior but cannot model target expiry.
 /// in Loom tests preserves lookup behavior but cannot model target expiry.
 /// Ordinary tests cover the expired-target delivery fallback.
 #[derive(Debug)]
 pub(crate) struct Weak<T>(Arc<T>);
-50
 impl<T> Weak<T> {
     /// Creates the modeled registry reference.
     pub(crate) fn from_arc(value: &Arc<T>) -> Self {
         Self(value.clone())
-55
+    }
-56

↧ Expand context

     /// Acquires an owning reference.
     pub(crate) fn upgrade(&self) -> Option<Arc<T>> {
         Some(self.0.clone())
-60
+    }
-61
+}
-62
 impl<T> Clone for Weak<T> {
     fn clone(&self) -> Self {
         Self(self.0.clone())
-66
+    }
-67
+}
-68
 /// Loom-backed mutex.
 #[derive(Debug)]
 pub(crate) struct Mutex<T>(loom::sync::Mutex<T>);
-72
 impl<T> Mutex<T> {
     /// Creates a mutex containing `value`.
     pub(crate) fn new(value: T) -> Self {
         Self(loom::sync::Mutex::new(value))

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/test_util/wire/connection.rs

@@ -76,76 +135,137 @@

↥ Expand context

-76
 impl Error for HarnessError {}
-78
 /// The identity assigned to an accepted connection.
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub struct ConnectionId(u64);
-82
 impl ConnectionId {
     /// Returns the numeric connection identity.
     pub fn as_u64(self) -> u64 {
         self.0
-87
+    }
-88
+}
-89
 impl fmt::Display for ConnectionId {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         self.0.fmt(f)
-93
+    }
-94
+}
-95

 /// Why a connection task stopped.
 #[non_exhaustive]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum ConnectionCloseReason {
     /// The peer closed the connection.
     ClientClosed,
     /// The script reached its end or explicitly closed the connection.
     ScriptCompleted,
     /// The script reset the connection.
     Reset,
     /// The harness dropped the transport without a protocol close signal.
     ScriptedTransportAbort,
     /// The harness was shut down.
     HarnessShutdown,
     /// The script failed.
     ScriptFailed,
-112
+}
-113
 /// An event recorded by the harness.
 #[non_exhaustive]
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub enum ConnectionEvent {

↧ Expand context

     /// A DNS lookup was performed.
     DnsLookup {
         /// The requested hostname.
         hostname: String,
     },
     /// A TCP connection was accepted.
     TcpAccepted {
         /// The accepted connection's identity.
         connection_id: ConnectionId,
         /// The endpoint that accepted the connection.
         endpoint_addr: SocketAddr,
     },
     /// A complete HTTP/1 request was received.
     Http1Request {
         /// The connection carrying the request.
         connection_id: ConnectionId,
         /// The endpoint that accepted the connection.
         endpoint_addr: SocketAddr,
         /// The request method.
         method: String,

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/tests/common/client.rs

@@ -1,1 +39,39 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Client configuration and request helpers shared by connection behavior tests.
-7
 use aws_smithy_async::rt::sleep::{SharedAsyncSleep, TokioSleep};
 use aws_smithy_async::time::SystemTimeSource;
 use aws_smithy_async::time::{SystemTimeSource, TimeSource};
 use aws_smithy_http_client::pool::{Client as PoolClient, ConnectionPool};
 use aws_smithy_http_client::proxy::ProxyConfig;
 #[cfg(any(feature = "__rustls", feature = "s2n-tls"))]
 use aws_smithy_http_client::tls;
 use aws_smithy_http_client::{Builder, Connector};
 use aws_smithy_runtime_api::client::dns::SharedDnsResolver;
 use aws_smithy_runtime_api::client::http::{
     http_client_fn, HttpClient, HttpConnector, HttpConnectorSettings, SharedHttpClient,
     SharedHttpConnector,
 };

↧ Expand context

 use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_runtime_api::client::runtime_components::{
     RuntimeComponents, RuntimeComponentsBuilder,
 };
 use http_body_util::BodyExt;
 use std::time::Duration;
-27
 /// Default timeout for test waits and deadline assertions.
 pub(crate) const WAIT: Duration = Duration::from_secs(5);
-30
 /// Settings applied uniformly by backend-neutral connection contracts.
 #[derive(Clone, Debug, Default)]
 pub(crate) struct BackendConfig {
     pub(crate) pool_idle_timeout: Option<Duration>,
     pub(crate) proxy_config: Option<ProxyConfig>,
     pub(crate) dns_resolver: Option<SharedDnsResolver>,
-37
+}
-38
 /// Hyper 1.x through `hyper_util::client::legacy::Client`.

@@ -230,230 +289,303 @@

↥ Expand context

             builder = builder.proxy_config(proxy_config);
-231
+        }
         let pool = builder
             .tls_provider(provider)
             .tls_context(tls_context)
             .build_https()
             .expect("valid connection-pool config");
         SharedHttpClient::new(PoolClient::new(&pool).expect("anonymous partition exists"))
-238
+    }
-239
+}
-240
 pub(crate) fn runtime_components() -> RuntimeComponents {
     RuntimeComponentsBuilder::for_tests()
         .with_time_source(Some(SystemTimeSource::new()))
         .with_sleep_impl(Some(SharedAsyncSleep::new(TokioSleep::new())))
         .build()
         .expect("valid runtime components")
-247
+}
-248
 pub(crate) fn connector_with_settings(

     client: &SharedHttpClient,
     settings: HttpConnectorSettings,
 ) -> SharedHttpConnector {
     client.http_connector(&settings, &runtime_components())
-254
+}
-255
 pub(crate) fn connector(client: &SharedHttpClient) -> SharedHttpConnector {
     connector_with_settings(client, HttpConnectorSettings::builder().build())
-258
+}
-259
 /// Builds a connector whose request telemetry uses `time_source`.
 #[allow(dead_code)]
 pub(crate) fn connector_with_time_source(
     client: &SharedHttpClient,
     time_source: impl TimeSource + 'static,
 ) -> SharedHttpConnector {
     let components = RuntimeComponentsBuilder::for_tests()
         .with_time_source(Some(time_source))
         .with_sleep_impl(Some(SharedAsyncSleep::new(TokioSleep::new())))
         .build()
         .expect("valid runtime components");
     client.http_connector(&HttpConnectorSettings::builder().build(), &components)
-+
+}
-+
 pub(crate) async fn send_request(
     connector: &SharedHttpConnector,
     request: HttpRequest,
 ) -> Result<HttpResponse, ConnectorError> {
     tokio::time::timeout(WAIT, connector.call(request))
         .await
         .expect("request should finish within the outer deadline")
-281
+}
-282
 pub(crate) async fn send_and_collect(

↧ Expand context

     connector: &SharedHttpConnector,
     request: HttpRequest,
 ) -> (u16, Vec<u8>) {
     let response = send_request(connector, request)
         .await
         .expect("request should succeed");
     collect_response(response).await
-291
+}
-292
 pub(crate) async fn get_and_collect(connector: &SharedHttpConnector, url: &str) -> (u16, Vec<u8>) {
     send_and_collect(
         connector,
         HttpRequest::get(url).expect("valid HTTP request"),
-297
+    )
     .await
-299
+}
-300
 pub(crate) async fn collect_response(response: HttpResponse) -> (u16, Vec<u8>) {
     let status = response.status().as_u16();
     let body = tokio::time::timeout(WAIT, response.into_body().collect())

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/tests/common/h2.rs

@@ -171,171 +293,323 @@

↥ Expand context

-171
+    }
-172
+}
-173
 #[derive(Clone, Debug)]
 pub(crate) struct H2Response {
     status: StatusCode,
     body: H2BodyPlan,
-178
+}
-179
 impl H2Response {
     pub(crate) fn new(status: StatusCode) -> Self {
         Self {
             status,
             body: H2BodyPlan::complete(Bytes::new()),
-185
+        }
-186
+    }
-187
     pub(crate) fn ok(body: impl Into<Bytes>) -> Self {
         Self::new(StatusCode::OK).body(H2BodyPlan::complete(body))
-190
+    }

-191
     pub(crate) fn body(mut self, body: H2BodyPlan) -> Self {
         self.body = body;
         self
-195
+    }
-196
+}
-197
 #[derive(Clone, Debug)]
 pub(crate) enum H2StreamScript {
     Respond(H2Response),
     /// Waits at the response-head boundary before sending the response.
     RespondAfter {
         response: H2Response,
         send: GateWaiter,
     },
     /// Sends the response before a gate permits the request body to be drained.
     RespondBeforeReceivingRequestBody {
         response: H2Response,
         receive: GateWaiter,
     },
     Reset(Reason),
-212
+}
-213
 impl H2StreamScript {
     pub(crate) fn respond(response: H2Response) -> Self {
         Self::Respond(response)
-217
+    }
-218
     /// Delays the response head until the gate is released.
     pub(crate) fn respond_after(response: H2Response, send: GateWaiter) -> Self {
         Self::RespondAfter { response, send }
-+
+    }
-+
     /// Delays request-body reads until after the response and gate release.
     pub(crate) fn respond_before_receiving_request_body(
         response: H2Response,
         receive: GateWaiter,
     ) -> Self {
         Self::RespondBeforeReceivingRequestBody { response, receive }
-230
+    }
-231
     pub(crate) fn reset(reason: Reason) -> Self {
         Self::Reset(reason)
-234
+    }
-235
+}
-236
 #[derive(Clone, Debug, Default)]
 pub(crate) struct H2ConnectionScript {
     routes: HashMap<String, H2StreamScript>,
     fallback: Option<H2StreamScript>,
     allow_handshake_abandonment: bool,
     goaway_on_ready: bool,
     client_close_behavior: ClientCloseBehavior,
-+
+}
-+
 /// How a scripted connection handles active stream tasks after client close.
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
 enum ClientCloseBehavior {
     /// Wait for stream scripts and report their failures.
     #[default]
     Drain,
     /// Abort scripts that a test deliberately leaves blocked.
     AbortStreams,
-254
+}
-255
 impl H2ConnectionScript {
     pub(crate) fn new() -> Self {
         Self::default()
-259
+    }
-260
     pub(crate) fn route(mut self, path: impl Into<String>, script: H2StreamScript) -> Self {
         let path = path.into();
         assert!(
             self.routes.insert(path.clone(), script).is_none(),
             "duplicate H2 route {path:?}"
         );
         self
-268
+    }
-269
     pub(crate) fn fallback(mut self, script: H2StreamScript) -> Self {
         self.fallback = Some(script);
         self
-273
+    }
-274
     pub(crate) fn allow_handshake_abandonment(mut self) -> Self {
         self.allow_handshake_abandonment = true;
         self
-278
+    }
-279
     pub(crate) fn goaway_on_ready(mut self) -> Self {
         self.goaway_on_ready = true;
         self
-283
+    }
-284
     /// Aborts active stream scripts when the client closes the connection.
     ///
     /// Use this only when the test intentionally strands a stream while
     /// terminating the runtime that owns its transport.
     pub(crate) fn abort_streams_on_client_close(mut self) -> Self {
         self.client_close_behavior = ClientCloseBehavior::AbortStreams;
         self
-+
+    }
-+
     fn script_for(&self, path: &str) -> Option<H2StreamScript> {
         self.routes.get(path).or(self.fallback.as_ref()).cloned()
-296
+    }
-297
+}
-298
 #[derive(Clone, Debug)]
 pub(crate) enum H2ConnectionPlan {
     Queue(VecDeque<H2ConnectionScript>),
     Unbounded(H2ConnectionScript),
-303
+}

↧ Expand context

-304
 impl H2ConnectionPlan {
     pub(crate) fn queue(scripts: impl IntoIterator<Item = H2ConnectionScript>) -> Self {
         Self::Queue(scripts.into_iter().collect())
-308
+    }
-309
     pub(crate) fn unbounded(script: H2ConnectionScript) -> Self {
         Self::Unbounded(script)
-312
+    }
-313
     fn next_script(&mut self) -> Option<H2ConnectionScript> {
         match self {
             Self::Queue(scripts) => scripts.pop_front(),
             Self::Unbounded(script) => Some(script.clone()),
-318
+        }
-319
+    }
-320
+}
-321
 #[derive(Debug)]
 struct RecordedState {

@@ -368,398 +427,458 @@

↥ Expand context

         control: mpsc::UnboundedSender<ConnectionCommand>,
     ) {
         self.controls
             .lock()
             .unwrap_or_else(|err| err.into_inner())
             .insert(connection_id, control);
-404
+    }
-405
     fn remove_control(&self, connection_id: H2ConnectionId) {
         self.controls
             .lock()
             .unwrap_or_else(|err| err.into_inner())
             .remove(&connection_id);
-411
+    }
-412
     fn control(
         &self,
         connection_id: H2ConnectionId,
     ) -> Option<mpsc::UnboundedSender<ConnectionCommand>> {
         self.controls

             .lock()
             .unwrap_or_else(|err| err.into_inner())
             .get(&connection_id)
             .cloned()
-422
+    }
-423
+}
-424
 #[derive(Debug)]
 enum ConnectionCommand {
     GracefulShutdown(oneshot::Sender<()>),
     AbortTransport(oneshot::Sender<()>),
-429
+}
-430
 #[derive(Debug, Default)]
 pub(crate) struct H2ServerBuilder {
     connections: Option<H2ConnectionPlan>,
-434
+}
-435
 impl H2ServerBuilder {
     pub(crate) fn connections(mut self, connections: H2ConnectionPlan) -> Self {
         self.connections = Some(connections);

↧ Expand context

         self
-440
+    }
-441
     pub(crate) async fn start(self) -> Result<H2TestServer, H2HarnessError> {
         let connections = self
             .connections
             .ok_or_else(|| H2HarnessError::new("an H2 connection plan is required"))?;
         let tls_acceptor = tls::SERVER_IDENTITY
             .acceptor(&[b"h2"])
             .map_err(|err| H2HarnessError::new(format!("failed to configure TLS: {err}")))?;
         let listener = TcpListener::bind("127.0.0.1:0")
             .await
             .map_err(|err| H2HarnessError::new(format!("failed to bind H2 listener: {err}")))?;
         let listen_addr = listener.local_addr().map_err(|err| {
             H2HarnessError::new(format!("failed to read H2 listener address: {err}"))
         })?;
         let state = Arc::new(SharedState::new());
         let (shutdown, shutdown_rx) = watch::channel(false);
         let listener_state = state.clone();
         let listener_task = tokio::spawn(run_listener(

@@ -507,538 +566,625 @@

↥ Expand context

-538
     pub(crate) async fn send_graceful_goaway(
         &self,
         connection_id: H2ConnectionId,
     ) -> Result<(), H2HarnessError> {
         let control = self.state.control(connection_id).ok_or_else(|| {
             H2HarnessError::new(format!("H2 connection {connection_id:?} is not active"))
         })?;
         let (acknowledged, ack) = oneshot::channel();
         control
             .send(ConnectionCommand::GracefulShutdown(acknowledged))
             .map_err(|_| {
                 H2HarnessError::new(format!(
                     "failed to send graceful shutdown to H2 connection {connection_id:?}"
                 ))
             })?;
         tokio::time::timeout(WAIT, ack)
             .await
             .map_err(|_| {
                 H2HarnessError::new(format!(

                     "timed out waiting for H2 connection {connection_id:?} to start graceful shutdown"
                 ))
             })?
             .map_err(|_| {
                 H2HarnessError::new(format!(
                     "H2 connection {connection_id:?} closed before graceful shutdown started"
                 ))
             })
-566
+    }
-567
     /// Drops one active transport without sending GOAWAY.
     pub(crate) async fn abort_transport(
         &self,
         connection_id: H2ConnectionId,
     ) -> Result<(), H2HarnessError> {
         let control = self.state.control(connection_id).ok_or_else(|| {
             H2HarnessError::new(format!("H2 connection {connection_id:?} is not active"))
         })?;
         let (acknowledged, ack) = oneshot::channel();
         control
             .send(ConnectionCommand::AbortTransport(acknowledged))
             .map_err(|_| {
                 H2HarnessError::new(format!("failed to abort H2 transport {connection_id:?}"))
             })?;
         tokio::time::timeout(WAIT, ack)
             .await
             .map_err(|_| {
                 H2HarnessError::new(format!(
                     "timed out waiting to abort H2 transport {connection_id:?}"
                 ))
             })?
             .map_err(|_| {
                 H2HarnessError::new(format!(
                     "H2 connection {connection_id:?} closed before its transport was aborted"
                 ))
             })
-+
+    }
-+
     pub(crate) async fn shutdown(mut self) -> Result<(), H2HarnessError> {
         self.shutdown.send_replace(true);
         let mut listener_task = self
             .listener_task
             .take()
             .expect("listener task should exist until shutdown");
         match tokio::time::timeout(WAIT, &mut listener_task).await {
             Ok(Ok(())) => {}
             Ok(Err(err)) => {
                 return Err(H2HarnessError::new(format!(

↧ Expand context

                     "H2 listener task failed: {err}"
                 )))
-608
+            }
             Err(_) => {
                 listener_task.abort();
                 return Err(H2HarnessError::new(
                     "timed out waiting for H2 listener and connection tasks to stop",
                 ));
-614
+            }
-615
+        }
         if let Some(failure) = self.state.failure() {
             return Err(failure);
-618
+        }
         Ok(())
-620
+    }
-621
+}
-622
 impl Drop for H2TestServer {
     fn drop(&mut self) {
         self.shutdown.send_replace(true);

@@ -711,770 +791,856 @@

↥ Expand context

         connection_id,
         alpn: alpn.clone(),
     });
     if alpn.as_deref() != Some(b"h2") {
         return Err(H2HarnessError::new(format!(
             "connection {connection_id:?} negotiated unexpected ALPN {alpn:?}"
         )));
-777
+    }
-778
     let observed = ObservedIo::new(tls_stream, connection_id, state.clone());
     let mut connection =
         match tokio::time::timeout(WAIT, h2::server::Builder::new().handshake(observed)).await {
             Err(_) => return Err(H2HarnessError::new("timed out waiting for H2 handshake")),
             Ok(Ok(connection)) => connection,
             Ok(Err(err)) if script.allow_handshake_abandonment => {
                 state.record_event(H2Event::H2HandshakeAbandoned {
                     connection_id,
                     error: err.to_string(),
                 });
                 return Ok(ConnectionCloseReason::ClientClosed);

-790
+            }
             Ok(Err(err)) => return Err(H2HarnessError::new(format!("H2 handshake failed: {err}"))),
         };
     state.record_event(H2Event::H2Ready { connection_id });
     if script.goaway_on_ready {
         connection.abrupt_shutdown(Reason::NO_ERROR);
-796
+    }
-797
     let mut stream_tasks = JoinSet::new();
     let mut shutting_down = false;
     let mut transport_aborted = false;
     let mut graceful_shutdown = script.goaway_on_ready;
     let mut control_open = true;
     // Tracks whether the connection ended because the client closed it (accept returned None)
     // vs. a scripted GOAWAY completing.
     let mut client_initiated_close = false;
-806
     loop {
         tokio::select! {
             changed = shutdown.changed(), if !shutting_down => {
                 if changed.is_err() || *shutdown.borrow() {
                     shutting_down = true;
                     connection.abrupt_shutdown(Reason::NO_ERROR);
-813
+                }
-814
+            }
             command = control.recv(), if control_open => {
                 match command {
                     Some(ConnectionCommand::GracefulShutdown(acknowledged)) => {
                         graceful_shutdown = true;
                         connection.graceful_shutdown();
                         let _ = acknowledged.send(());
-821
+                    }
                     Some(ConnectionCommand::AbortTransport(acknowledged)) => {
                         transport_aborted = true;
                         let _ = acknowledged.send(());
                         break;
-+
+                    }
                     None => control_open = false,
-828
+                }
-829
+            }
             accepted = connection.accept() => {
                 match accepted {
                     Some(Ok((request, mut respond))) => {
                         let stream_id = respond.stream_id().as_u32();
                         let method = request.method().clone();
                         let path = request.uri().path().to_string();
                         let proxy_authorization = request

↧ Expand context

                             .headers()
                             .get(http_1x::header::PROXY_AUTHORIZATION)
                             .map(|value| value.to_str().unwrap_or_default().to_string());
                         state.record_event(H2Event::StreamAccepted {
                             connection_id,
                             stream_id,
                             method,
                             path: path.clone(),
                             proxy_authorization,
                         });
                         let Some(stream_script) = script.script_for(&path) else {
                             respond.send_reset(Reason::PROTOCOL_ERROR);
                             return Err(H2HarnessError::new(format!(
                                 "connection {connection_id:?} received unscripted H2 path {path:?}"
                             )));
                         };
                         let held_request = match &stream_script {
                             H2StreamScript::RespondBeforeReceivingRequestBody { receive, .. } => {
                                 Some((request.into_body(), receive.clone()))
-856
+                            }

@@ -808,873 +923,1025 @@

↥ Expand context

                             connection_id,
                             stream_id,
                             stream_script,
                             respond,
                             state.clone(),
                             held_request,
                         ));
-880
+                    }
                     Some(Err(err)) => {
                         if shutting_down
                             || (graceful_shutdown && err.reason() == Some(Reason::NO_ERROR))
-884
+                        {
                             break;
-886
+                        }
                         return Err(H2HarnessError::new(format!(
                             "H2 connection {connection_id:?} failed while accepting a stream: {err}"
                         )));
-890
+                    }
                     None => {
                         client_initiated_close = true;

                         break;
-894
+                    }
-895
+                }
-896
+            }
             completed = stream_tasks.join_next(), if !stream_tasks.is_empty() => {
                 record_stream_task_result(&state, completed);
-899
+            }
-900
+        }
-901
+    }
-902
     if shutting_down {
     let abort_streams = shutting_down
         || transport_aborted
         || (client_initiated_close
             && script.client_close_behavior == ClientCloseBehavior::AbortStreams);
     if abort_streams {
         stream_tasks.abort_all();
         while stream_tasks.join_next().await.is_some() {}
         Ok(ConnectionCloseReason::HarnessShutdown)
         Ok(aborted_connection_close_reason(
             transport_aborted,
             shutting_down,
             client_initiated_close,
         ))
     } else {
         let drain = async {
             while let Some(completed) = stream_tasks.join_next().await {
                 record_stream_task_result(&state, Some(completed));
-919
+            }
         };
         if tokio::time::timeout(WAIT, drain).await.is_err() {
             state.record_failure(H2HarnessError::new(format!(
                 "H2 connection {connection_id:?} closed before its stream scripts completed"
             )));
             stream_tasks.abort_all();
             while stream_tasks.join_next().await.is_some() {}
-927
+        }
         if client_initiated_close {
             Ok(ConnectionCloseReason::ClientClosed)
         } else {
             Ok(ConnectionCloseReason::ScriptCompleted)
-932
+        }
-933
+    }
-934
+}
-935
 fn aborted_connection_close_reason(
     transport_aborted: bool,
     shutting_down: bool,
     client_initiated_close: bool,
 ) -> ConnectionCloseReason {
     debug_assert!(transport_aborted || shutting_down || client_initiated_close);
     if transport_aborted {
         ConnectionCloseReason::ScriptedTransportAbort
     } else if shutting_down {
         ConnectionCloseReason::HarnessShutdown
     } else {
         ConnectionCloseReason::ClientClosed
-+
+    }
-+
+}
-+
 #[test]
 fn harness_shutdown_precedes_client_close() {
     assert_eq!(
         ConnectionCloseReason::HarnessShutdown,
         aborted_connection_close_reason(false, true, true,)
     );
-+
+}
-+
 fn record_stream_task_result(
     state: &SharedState,
     completed: Option<Result<Result<(), H2HarnessError>, tokio::task::JoinError>>,
 ) {
     match completed {
         Some(Ok(Ok(()))) | None => {}
         Some(Ok(Err(err))) => state.record_failure(err),
         Some(Err(err)) => {
             state.record_failure(H2HarnessError::new(format!("H2 stream task failed: {err}")))
-968
+        }
-969
+    }
-970
+}
-971
 async fn run_stream(
     connection_id: H2ConnectionId,
     stream_id: u32,
     script: H2StreamScript,
     mut respond: SendResponse<Bytes>,
     state: Arc<SharedState>,
     held_request: Option<(RecvStream, GateWaiter)>,
 ) -> Result<(), H2HarnessError> {
     if let H2StreamScript::RespondAfter { send, .. } = &script {
         send.wait()
             .await
             .map_err(|err| H2HarnessError::new(format!("H2 response-head gate failed: {err}")))?;
-+
+    }
     match script {
         H2StreamScript::Reset(reason) => {
             respond.send_reset(reason);
             state.record_event(H2Event::ResetSent {
                 connection_id,
                 stream_id,
                 reason,
             });
-993
+        }
         H2StreamScript::Respond(response)
         | H2StreamScript::RespondAfter { response, send: _ }
         | H2StreamScript::RespondBeforeReceivingRequestBody {
             response,
             receive: _,
         } => {
             let end_stream =
                 matches!(&response.body, H2BodyPlan::Complete(body) if body.is_empty());
             let response_head = Response::builder()
                 .status(response.status)
                 .body(())
                 .map_err(|err| {

↧ Expand context

                     H2HarnessError::new(format!("failed to build H2 response: {err}"))
                 })?;
             let mut send = respond
                 .send_response(response_head, end_stream)
                 .map_err(|err| H2HarnessError::new(format!("failed to send H2 response: {err}")))?;
             if end_stream {
                 state.record_event(H2Event::ResponseCompleted {
                     connection_id,
                     stream_id,
                 });
                 receive_request_body(held_request).await?;
                 return Ok(());
-1018
+            }
-1019
             match response.body {
                 H2BodyPlan::Complete(body) => {
                     send.send_data(body, true).map_err(|err| {
                         H2HarnessError::new(format!("failed to send H2 response body: {err}"))
                     })?;
                     state.record_event(H2Event::ResponseCompleted {

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/tests/common/runtime.rs

@@ -0,1 +0,122 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-+
 //! Independently driven Tokio runtimes for partition-placement tests.
-+
 use aws_smithy_http_client::pool::DriverSpawner;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{mpsc, Arc};
 use std::thread::{self, JoinHandle};
 use tokio::runtime::{Handle, Id};
 use tokio::sync::oneshot;
-+
 /// A current-thread Tokio runtime driven by its own operating-system thread.
 #[derive(Debug)]
 pub(crate) struct DrivenRuntime {
     handle: Handle,
     runtime_id: Id,
     shutdown: Option<oneshot::Sender<()>>,
     thread: Option<JoinHandle<()>>,
     submitted_tasks: Arc<AtomicUsize>,
-+
+}
-+
 impl DrivenRuntime {
     /// Starts a current-thread runtime and waits until its handle is available.
     pub(crate) fn start(name: &str) -> Self {
         let (ready_tx, ready_rx) = mpsc::sync_channel(1);
         let (shutdown, shutdown_rx) = oneshot::channel();
         let submitted_tasks = Arc::new(AtomicUsize::new(0));
         let thread = thread::Builder::new()
             .name(name.to_owned())
             .spawn(move || {
                 let runtime = tokio::runtime::Builder::new_current_thread()
                     .enable_all()
                     .build()
                     .expect("test runtime should build");
                 let handle = runtime.handle().clone();
                 let runtime_id = handle.id();
                 ready_tx
                     .send((handle, runtime_id))
                     .expect("runtime owner should receive its handle");
                 runtime.block_on(async {
                     let _ = shutdown_rx.await;
                 });
             })
             .expect("test runtime thread should start");
         let (handle, runtime_id) = ready_rx
             .recv()
             .expect("test runtime should publish its handle");
         Self {
             handle,
             runtime_id,
             shutdown: Some(shutdown),
             thread: Some(thread),
             submitted_tasks,
-+
+        }
-+
+    }
-+
     /// Returns a spawner that records task placement on this runtime.
     pub(crate) fn driver_spawner(&self) -> RuntimeDriverSpawner {
         RuntimeDriverSpawner {
             handle: self.handle.clone(),
             submitted_tasks: self.submitted_tasks.clone(),
-+
+        }
-+
+    }
-+
     /// Spawns test orchestration work on this runtime.
     pub(crate) fn spawn<F>(&self, future: F) -> tokio::task::JoinHandle<F::Output>
     where
         F: Future + Send + 'static,
         F::Output: Send + 'static,
-+
+    {
         self.handle.spawn(future)
-+
+    }
-+
     /// Returns the runtime's stable Tokio identity.
     pub(crate) fn id(&self) -> Id {
         self.runtime_id
-+
+    }
-+
     /// Returns how many connection-owned tasks were submitted to this runtime.
     pub(crate) fn submitted_tasks(&self) -> usize {
         self.submitted_tasks.load(Ordering::SeqCst)
-+
+    }
-+
     /// Stops the runtime, drops all runtime-owned tasks, and joins its thread.
     pub(crate) fn shutdown(mut self) {
         self.stop();
-+
+    }
-+
     fn stop(&mut self) {
         if let Some(shutdown) = self.shutdown.take() {
             let _ = shutdown.send(());
-+
+        }
         if let Some(thread) = self.thread.take() {
             thread.join().expect("test runtime thread should not panic");
-+
+        }
-+
+    }
-+
+}
-+
 impl Drop for DrivenRuntime {
     fn drop(&mut self) {
         self.stop();
-+
+    }
-+
+}
-+
 /// Places connection-owned tasks on one [`DrivenRuntime`].
 #[derive(Clone, Debug)]
 pub(crate) struct RuntimeDriverSpawner {
     handle: Handle,
     submitted_tasks: Arc<AtomicUsize>,
-+
+}
-+
 impl DriverSpawner for RuntimeDriverSpawner {
     fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
         self.submitted_tasks.fetch_add(1, Ordering::SeqCst);
         drop(self.handle.spawn(driver));
-+
+    }
-+
+}

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/tests/connection_pool_test.rs

@@ -1,1 +66,81 @@

↥ Expand context

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Behavior specific to the partition-aware connection pool.
-7
 #![cfg(all(

     feature = "wire-mock",
     feature = "default-client",
     feature = "rt-tokio"
 ))]
-13
 mod common {
     #[allow(dead_code)]
     pub(crate) mod client;
-17
+}
-18
 #[path = "common/runtime.rs"]
 mod runtime;
-+
 use aws_smithy_async::test_util::ManualTimeSource;
 use aws_smithy_http_client::pool::{
     Client, ConnectPath, ConnectionEstablishmentId, ConnectionEstablishmentStage, ConnectionEvent,
     ConnectionPool, ConnectionProtocol, ConnectionReuseScope, OriginKey, Partition, PartitionId,
     TokioDriverSpawner,
 };
 use aws_smithy_http_client::test_util::wire::connection::{
     BodyPlan, ConnectionCloseReason, ConnectionEvent as WireConnectionEvent, ConnectionTestHarness,
     EndpointPlan, Http1Response, Http1Script, ManualGate,
     EndpointPlan, Http1Response, Http1Script, ManualGate, SocketScript,
 };
 use aws_smithy_runtime_api::client::connection::{CaptureSmithyConnection, ConnectionId};
 use aws_smithy_runtime_api::client::http::telemetry::{
     CaptureHttpAttemptTelemetry, ConnectionUsage,
 };
 use aws_smithy_runtime_api::client::connection::ConnectionId;
 use aws_smithy_runtime_api::client::http::{SharedHttpClient, SharedHttpConnector};
 use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
 use aws_smithy_types::body::SdkBody;
 use bytes::Bytes;
 use common::client as test_client;
 use http_body_1x::{Body, Frame, SizeHint};
 use std::convert::Infallible;
 use std::future::Future;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use std::task::{Context, Poll};
 use std::time::{Duration, UNIX_EPOCH};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
-52
 const IP1: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
 const IP2: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2));
-55
 #[derive(Debug)]
 enum ObservedConnectionEvent {
     Failed {
         establishment: ConnectionEstablishmentId,
         origin: OriginKey,
         stage: ConnectionEstablishmentStage,

↧ Expand context

         remote_addr: Option<SocketAddr>,
         protocol: Option<ConnectionProtocol>,
         error: String,
     },
     Opened {
         establishment: ConnectionEstablishmentId,
         connection: aws_smithy_runtime_api::client::connection::ConnectionId,
         establishment_partition: PartitionId,
         establishment_origin: OriginKey,
         connection_origin: OriginKey,
         connection_partition: PartitionId,
         remote_addr: Option<SocketAddr>,
         protocol: ConnectionProtocol,
         connect_path: ConnectPath,
         handshake_measured: bool,
     },
-78
+}
-79
 /// Resolves one declared partition through the public Smithy client boundary.
 fn shared_client(pool: &ConnectionPool, partition: PartitionId) -> SharedHttpClient {

@@ -292,307 +407,590 @@

↥ Expand context

-307
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             Http1Script::responses([Http1Response::ok().body("custom dns")]),
-312
+        )
         .dns(HOST, [IP1])
         .build()
         .await
         .expect("harness should start");
     let partition = PartitionId::from_index(1);
     let pool = ConnectionPool::builder()
         .partitions([Partition::new(partition, TokioDriverSpawner::current())])
         .dns_resolver(harness.dns_resolver())
         .build_http()
         .expect("valid pool");
     let client = shared_client(&pool, partition);
     let connector = connector(&client);
     let url = format!("http://{HOST}:{}/custom-dns", harness.port());
-326

     let (status, body) = test_client::get_and_collect(&connector, &url).await;
     assert_eq!((status, body.as_slice()), (200, b"custom dns".as_slice()));
     assert_eq!(1, harness.dns_lookup_count());
-330
     drop(connector);
     drop(client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-335
+}
-336
 #[tokio::test]
 async fn h1_attempt_telemetry_distinguishes_fresh_and_reused_connections() {
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             Http1Script::responses([
                 Http1Response::ok().body("first"),
                 Http1Response::ok().body("second"),
             ]),
-+
+        )
         .build()
         .await
         .expect("harness should start");
     let opened = Arc::new(Mutex::new(None));
     let pool = ConnectionPool::builder()
         .event_listener({
             let opened = opened.clone();
             move |event: &ConnectionEvent<'_>| {
                 if let ConnectionEvent::Opened(event) = event {
                     *opened.lock().unwrap() =
                         Some((event.connection().id(), event.stats().total_duration()));
-+
+                }
-+
+            }
         })
         .build_http()
         .expect("valid pool");
     let client = SharedHttpClient::new(Client::new(&pool).expect("anonymous partition"));
     let connector = connector(&client);
-+
     let first_connection_capture = CaptureSmithyConnection::new();
     let first_capture = CaptureHttpAttemptTelemetry::new();
     let mut first = HttpRequest::get(harness.endpoint_url()).expect("valid request");
     first.add_extension(first_connection_capture.clone());
     first.add_extension(first_capture.clone());
     test_client::send_and_collect(&connector, first).await;
-+
     let second_capture = CaptureHttpAttemptTelemetry::new();
     let mut second = HttpRequest::get(harness.endpoint_url()).expect("valid request");
     second.add_extension(second_capture.clone());
     test_client::send_and_collect(&connector, second).await;
-+
     let first = first_capture.get();
     let second = second_capture.get();
     assert!(first.connector_call_duration().is_some());
     assert!(second.connector_call_duration().is_some());
     assert_eq!(
         first.acquisition().expect("first acquisition").usage(),
         ConnectionUsage::Fresh
     );
     assert_eq!(
         second.acquisition().expect("second acquisition").usage(),
         ConnectionUsage::Reused
     );
     let first_connection = first.connection().expect("first connection metadata");
     let second_connection = second.connection().expect("second connection metadata");
     let captured_connection = first_connection_capture
         .get()
         .expect("connection capture should contain selected metadata");
     assert_eq!(
         captured_connection.connection_id(),
         first_connection.connection_id()
     );
     assert_eq!(
         captured_connection.establishment(),
         first_connection.establishment(),
         "connection and attempt captures must observe the same selected connection"
     );
     assert_eq!(
         first_connection.connection_id(),
         second_connection.connection_id()
     );
     assert_eq!(
         first_connection.establishment(),
         second_connection.establishment(),
         "reuse must retain the establishment that created the connection"
     );
     let first_establishment = first_connection
         .establishment()
         .expect("first connection establishment");
     let (opened_connection, opened_duration) =
         opened.lock().unwrap().expect("opened connection event");
     assert_eq!(
         first_connection.connection_id(),
         Some(opened_connection),
         "request telemetry and lifecycle events must identify the same connection"
     );
     assert_eq!(
         first_establishment.total_duration(),
         opened_duration,
         "request metadata and the opened event must share one establishment measurement"
     );
     assert_eq!(1, harness.tcp_accepted_count());
-+
     drop(connector);
     drop(client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-+
+}
-+
 #[tokio::test]
 async fn acquisition_duration_ends_when_hyper_accepts_the_request() {
     let response_gate = ManualGate::new();
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             SocketScript::new()
                 .read_http1_request()
                 .wait(response_gate.waiter())
                 .write_all(
                     "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: keep-alive\r\n\r\nok",
-+
+                )
                 .await_client_close(),
-+
+        )
         .build()
         .await
         .expect("harness should start");
     let pool_time = ManualTimeSource::new(UNIX_EPOCH);
     let attempt_time = ManualTimeSource::new(UNIX_EPOCH);
     let pool = ConnectionPool::builder()
         .time_source(pool_time)
         .build_http()
         .expect("valid pool");
     let client = SharedHttpClient::new(Client::new(&pool).expect("anonymous partition"));
     let connector = test_client::connector_with_time_source(&client, attempt_time.clone());
     let capture = CaptureHttpAttemptTelemetry::new();
     let mut request = HttpRequest::get(harness.endpoint_url()).expect("valid request");
     request.add_extension(capture.clone());
-+
     let send = tokio::spawn({
         let connector = connector.clone();
         async move { test_client::send_and_collect(&connector, request).await }
     });
     response_gate
         .wait_until_reached(test_client::WAIT)
         .await
         .expect("server should receive the accepted request");
     attempt_time.advance(Duration::from_secs(10));
     response_gate.release();
     let (status, body) = send.await.expect("request task");
     assert_eq!((status, body.as_slice()), (200, b"ok".as_slice()));
-+
     let telemetry = capture.get();
     assert_eq!(
         telemetry.acquisition().expect("acquisition").duration(),
         Some(Duration::ZERO)
     );
     assert_eq!(
         telemetry.connector_call_duration(),
         Some(Duration::from_secs(10))
     );
-+
     drop(connector);
     drop(client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-+
+}
-+
 #[tokio::test]
 async fn bounded_waiter_proceeds_after_the_active_h1_returns() {
     let body_gate = ManualGate::new();
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             Http1Script::responses([
                 Http1Response::ok().body_plan(BodyPlan::split_at_gate(
                     "first-",
                     body_gate.waiter(),
                     "body",
                 )),
                 Http1Response::ok().body("second"),
             ]),
-508
+        )
         .build()
         .await
         .expect("harness should start");
     let attempt_time = ManualTimeSource::new(UNIX_EPOCH);
     let pool = ConnectionPool::builder()
         .max_connections_per_host(1)
         .build_http()
         .expect("valid pool");
     let client = SharedHttpClient::new(Client::new(&pool).expect("anonymous partition"));
     let connector = connector(&client);
     let connector = test_client::connector_with_time_source(&client, attempt_time.clone());
-519
     let first = test_client::send_request(
         &connector,
         HttpRequest::get(harness.endpoint_url()).expect("valid request"),
-523
+    )
     .await
     .expect("first request should reach response headers");
     body_gate
         .wait_until_reached(test_client::WAIT)
         .await
         .expect("first response should reach its body gate");
-530
     let second_capture = CaptureHttpAttemptTelemetry::new();
     let mut second_request = HttpRequest::get(harness.endpoint_url()).expect("valid request");
     second_request.add_extension(second_capture.clone());
     let second_connector = connector.clone();
     let second_url = harness.endpoint_url();
     let mut second =
         tokio::spawn(
             async move { test_client::get_and_collect(&second_connector, &second_url).await },
         );
     let mut second = tokio::spawn(async move {
         test_client::send_and_collect(&second_connector, second_request).await
     });
     assert!(
         tokio::time::timeout(Duration::from_millis(100), &mut second)
             .await
             .is_err(),
         "the second request must wait while the only admitted H1 is active"
     );
     assert_eq!(1, harness.tcp_accepted_count());
-545
     attempt_time.advance(Duration::from_secs(5));
     body_gate.release();
     let (status, body) = test_client::collect_response(first).await;
     assert_eq!((status, body.as_slice()), (200, b"first-body".as_slice()));
     let (status, body) = second.await.expect("second request task should not panic");
     assert_eq!((status, body.as_slice()), (200, b"second".as_slice()));
     assert_eq!(1, harness.tcp_accepted_count());
     let telemetry = second_capture.get();
     let acquisition = telemetry.acquisition().expect("second acquisition");
     assert_eq!(acquisition.duration(), Some(Duration::from_secs(5)));
     assert_eq!(acquisition.usage(), ConnectionUsage::Reused);
     assert_eq!(
         telemetry.connector_call_duration(),
         Some(Duration::from_secs(5))
     );
-561
     drop(connector);
     drop(client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-566
+}
-567
 #[tokio::test]
 async fn eligible_partition_borrows_the_peer_h1() {
     let harness = ConnectionTestHarness::builder()

↧ Expand context

         .endpoint(
             IP1,
             Http1Script::responses([
                 Http1Response::ok().body("first"),
                 Http1Response::ok().body("second"),
             ]),
-577
+        )
         .build()
         .await
         .expect("harness should start");
     let first_id = PartitionId::from_index(1);
     let second_id = PartitionId::from_index(2);
     let pool = ConnectionPool::builder()
         .partitions([
             Partition::new(first_id, TokioDriverSpawner::current()),
             Partition::new(second_id, TokioDriverSpawner::current()),
         ])
         .connection_reuse_scope(ConnectionReuseScope::Pool)
         .max_connections_per_host(1)
         .build_http()

@@ -410,593 +469,834 @@

↥ Expand context

     let second_client = shared_client(&pool, second_id);
     let first_connector = connector(&first_client);
     let second_connector = connector(&second_client);
-596
     test_client::get_and_collect(&first_connector, &harness.endpoint_url()).await;
     test_client::get_and_collect(&second_connector, &harness.endpoint_url()).await;
-599
     let request_connections = harness
         .events()
         .into_iter()
         .filter_map(|event| match event {
             WireConnectionEvent::Http1Request { connection_id, .. } => Some(connection_id),
             _ => None,
         })
         .collect::<Vec<_>>();
     assert_eq!(2, request_connections.len());
     assert_eq!(
         request_connections[0], request_connections[1],
         "eligible peer demand should borrow the existing H1"
     );

     assert_eq!(1, harness.tcp_accepted_count());
-614
     drop(first_connector);
     drop(second_connector);
     drop(first_client);
     drop(second_client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-621
+}
-622
 /// Request body that remains open until the test releases it.
 struct HeldUpload {
     finish: oneshot::Receiver<()>,
     complete: bool,
-+
+}
-+
 impl Body for HeldUpload {
     type Data = Bytes;
     type Error = Infallible;
-+
     fn poll_frame(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
         if self.complete {
             return Poll::Ready(None);
-+
+        }
         match Pin::new(&mut self.finish).poll(cx) {
             Poll::Ready(_) => {
                 self.complete = true;
                 Poll::Ready(None)
-+
+            }
             Poll::Pending => Poll::Pending,
-+
+        }
-+
+    }
-+
     fn is_end_stream(&self) -> bool {
         self.complete
-+
+    }
-+
     fn size_hint(&self) -> SizeHint {
         SizeHint::default()
-+
+    }
-+
+}
-+
 fn held_upload() -> (oneshot::Sender<()>, SdkBody) {
     let (finish, finished) = oneshot::channel();
-+
+    (
         finish,
         SdkBody::from_body_1_x(HeldUpload {
             finish: finished,
             complete: false,
         }),
-+
+    )
-+
+}
-+
 #[tokio::test]
 async fn peer_h1_reuse_survives_borrower_runtime_shutdown() {
     use runtime::DrivenRuntime;
-+
     let peer_response_gate = ManualGate::new();
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             SocketScript::new()
                 .read_http1_request()
                 .write_all(
                     b"HTTP/1.1 200 OK\r\n\
                       Content-Length: 5\r\n\
                       Connection: keep-alive\r\n\
                       \r\n\
                       owner",
-+
+                )
                 .read_until(b"\r\n\r\n", 16 * 1024)
                 .wait(peer_response_gate.waiter())
                 .write_all(
                     b"HTTP/1.1 200 OK\r\n\
                       Content-Length: 4\r\n\
                       Connection: keep-alive\r\n\
                       \r\n\
                       peer",
-+
+                )
                 .read_until(b"0\r\n\r\n", 1024)
                 .read_http1_request()
                 .write_all(
                     b"HTTP/1.1 200 OK\r\n\
                       Content-Length: 6\r\n\
                       Connection: keep-alive\r\n\
                       \r\n\
                       reused",
-+
+                )
                 .await_client_close(),
-+
+        )
         .build()
         .await
         .expect("harness should start");
     let first_id = PartitionId::from_index(1);
     let second_id = PartitionId::from_index(2);
     let first_runtime = DrivenRuntime::start("h1-owner-one");
     let second_runtime = DrivenRuntime::start("h1-owner-two");
     assert_ne!(first_runtime.id(), second_runtime.id());
     let pool = ConnectionPool::builder()
         .partitions([
             Partition::new(first_id, first_runtime.driver_spawner()),
             Partition::new(second_id, second_runtime.driver_spawner()),
         ])
         .connection_reuse_scope(ConnectionReuseScope::Pool)
         .max_connections_per_host(1)
         .build_http()
         .expect("valid pool");
     let first_client = shared_client(&pool, first_id);
     let second_client = shared_client(&pool, second_id);
     let first_connector = connector(&first_client);
     let second_connector = connector(&second_client);
     let url = harness.endpoint_url();
-+
     let first_request = first_connector.clone();
     let first_url = url.clone();
     let (status, body) = first_runtime
         .spawn(async move { test_client::get_and_collect(&first_request, &first_url).await })
         .await
         .expect("owner request task should not panic");
     assert_eq!((status, body.as_slice()), (200, b"owner".as_slice()));
-+
     let (finish_upload, upload) = held_upload();
     let mut peer_request = HttpRequest::new(upload);
     peer_request
         .set_uri(url.clone())
         .expect("valid HTTP request URI");
     peer_request.set_method("POST").expect("valid HTTP method");
     let peer_connector = second_connector.clone();
     let peer = second_runtime
         .spawn(async move { test_client::send_and_collect(&peer_connector, peer_request).await });
     peer_response_gate
         .wait_until_reached(test_client::WAIT)
         .await
         .expect("peer request should reach the response gate");
     let owner_tasks_before_peer_completion = first_runtime.submitted_tasks();
     let borrower_tasks_before_peer_completion = second_runtime.submitted_tasks();
     peer_response_gate.release();
     let (status, body) = peer.await.expect("peer request task should not panic");
     assert_eq!((status, body.as_slice()), (200, b"peer".as_slice()));
     assert_eq!(
         owner_tasks_before_peer_completion + 1,
         first_runtime.submitted_tasks(),
         "deferred H1 readiness must use the connection owner's spawner"
     );
     assert_eq!(
         borrower_tasks_before_peer_completion,
         second_runtime.submitted_tasks(),
         "the borrowing partition must not own readiness for a peer connection"
     );
     finish_upload
         .send(())
         .expect("peer upload should remain open until readiness placement is observed");
-+
     second_runtime.shutdown();
-+
     let reused_request = first_connector.clone();
     let reused_url = url.clone();
     let (status, body) = first_runtime
         .spawn(async move { test_client::get_and_collect(&reused_request, &reused_url).await })
         .await
         .expect("reused request task should not panic");
     assert_eq!((status, body.as_slice()), (200, b"reused".as_slice()));
-+
     let request_connections = harness
         .events()
         .into_iter()
         .filter_map(|event| match event {
             WireConnectionEvent::Http1Request { connection_id, .. } => Some(connection_id),
             _ => None,
         })
         .collect::<Vec<_>>();
     assert_eq!(2, request_connections.len());
     assert!(
         request_connections
             .iter()
             .all(|connection| *connection == request_connections[0]),
         "owner and peer requests should use one HTTP/1 connection"
     );
     assert_eq!(1, harness.tcp_accepted_count());
-+
     drop(first_connector);
     drop(second_connector);
     drop(first_client);
     drop(second_client);
     drop(pool);
     first_runtime.shutdown();
     harness.shutdown().await.expect("clean harness shutdown");
-+
+}
-+
 #[tokio::test]
 async fn ineligible_partition_reclaims_peer_capacity() {
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             EndpointPlan::queue([
                 Http1Script::responses([Http1Response::ok().body("first")]),
                 Http1Script::responses([Http1Response::ok().body("second")]),
             ]),
-814
+        )

↧ Expand context

         .build()
         .await
         .expect("harness should start");
     let first_id = PartitionId::from_index(1);
     let second_id = PartitionId::from_index(2);
     let pool = ConnectionPool::builder()
         .partitions([
             Partition::new(first_id, TokioDriverSpawner::current()),
             Partition::new(second_id, TokioDriverSpawner::current()),
         ])
         .connection_reuse_scope(ConnectionReuseScope::Partition)
         .max_connections_per_host(1)
         .build_http()
         .expect("valid pool");
     let first_client = shared_client(&pool, first_id);
     let second_client = shared_client(&pool, second_id);
     let first_connector = connector(&first_client);
     let second_connector = connector(&second_client);
-833
     test_client::get_and_collect(&first_connector, &harness.endpoint_url()).await;

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/tests/h2_connection_behavior_test.rs

@@ -1,1 +53,58 @@

↥ Expand context

  */
-5
 //! HTTP/2 connection behavior contracts.
 //!
 //! Each contract is implementation-neutral and has an explicit runner for every
 //! HTTP client backend expected to preserve that behavior.
-10
 #![cfg(all(feature = "wire-mock", feature = "rustls-aws-lc"))]
-12
 mod common;
 #[path = "common/runtime.rs"]
 mod runtime;
-16
 use aws_smithy_http_client::pool::{
     Client as PoolClient, ConnectionPool, ConnectionReuseScope, Partition, PartitionId,
     TokioDriverSpawner,
 };
 use aws_smithy_http_client::test_util::wire::connection::{ConnectionCloseReason, ManualGate};
 use aws_smithy_http_client::tls;
 use aws_smithy_runtime_api::client::connection::{
     CaptureSmithyConnection, ConnectionMetadata as SmithyConnectionMetadata,
 };
 use aws_smithy_runtime_api::client::http::telemetry::{
     CaptureHttpAttemptTelemetry, ConnectionUsage,
 };
 use aws_smithy_runtime_api::client::http::{SharedHttpClient, SharedHttpConnector};
 use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
 use bytes::Bytes;
 use common::client as test_client;
 use common::client::{
     BackendConfig, HttpsClientBackend, HyperUtilLegacyPool, PartitionedConnectionPool,
 };
 use common::h2::{
     H2BodyPlan, H2ConnectionId, H2ConnectionPlan, H2ConnectionScript, H2Event, H2Response,
     H2StreamScript, H2TestServer,

↧ Expand context

 };
 use common::tls as test_tls;
 use h2::Reason;
 use http_body_util::BodyExt;
 use std::error::Error;
-44
 fn rustls_aws_lc() -> tls::Provider {
     tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc)
-47
+}
-48
 fn h2_client_with_provider(
     backend: &dyn HttpsClientBackend,
     provider: tls::Provider,
 ) -> SharedHttpClient {
     backend.build_https(
         BackendConfig::default(),
         provider,
         test_tls::SERVER_IDENTITY.client_context(),
-57
+    )
-58
+}

@@ -248,253 +307,443 @@

↥ Expand context

                 .all(|connection_id| *connection_id == ready_connections[0]),
             "all cold-start requests should converge on one established H2 connection"
         );
         body_gate.release();
-257
         while let Some(result) = requests.join_next().await {
             let (status, body) = result.expect("request task should not panic");
             assert_eq!((status, body.as_slice()), (200, b"cold".as_slice()));
-261
+        }
         assert_eq!(server.stream_count(), 4);
-263
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-267
+    }
-268
     #[tokio::test]
     async fn test_concurrent_cold_start_converges_on_one_h2_connection_with_hyper_util_legacy_pool()
-271
+    {
         concurrent_cold_start_converges_on_one_h2_connection(&HyperUtilLegacyPool).await;

-273
+    }
-274
     #[tokio::test]
     async fn test_concurrent_cold_start_converges_on_one_h2_connection_with_partitioned_pool() {
         concurrent_cold_start_converges_on_one_h2_connection(&PartitionedConnectionPool).await;
-278
+    }
-279
+}
-280
 mod connection_metadata {
     use super::*;
     use aws_smithy_async::test_util::ManualTimeSource;
     use std::time::{Duration, UNIX_EPOCH};
-+
     async fn capture_attempt(
         connector: &SharedHttpConnector,
         url: &str,
     ) -> aws_smithy_runtime_api::client::http::telemetry::HttpAttemptTelemetry {
         let capture = CaptureHttpAttemptTelemetry::new();
         let mut request = HttpRequest::get(url).expect("valid HTTP request");
         request.add_extension(capture.clone());
         let (status, body) = test_client::send_and_collect(connector, request).await;
         assert_eq!((status, body.as_slice()), (200, b"ok".as_slice()));
         capture.get()
-+
+    }
-+
     async fn attempt_telemetry_matches_backend_capability(
         backend: &dyn HttpsClientBackend,
         captures_selection: bool,
     ) {
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([
                 H2ConnectionScript::new().fallback(H2StreamScript::respond(H2Response::ok("ok")))
             ]))
             .start()
             .await
             .expect("H2 server should start");
         let client = h2_client(backend);
         let connector = test_client::connector(&client);
-+
         let first = capture_attempt(&connector, &server.url("/first")).await;
         let second = capture_attempt(&connector, &server.url("/second")).await;
         assert!(first.connector_call_duration().is_some());
         assert!(second.connector_call_duration().is_some());
-+
         if captures_selection {
             assert_eq!(
                 first.acquisition().expect("first acquisition").usage(),
                 ConnectionUsage::Fresh
             );
             assert_eq!(
                 second.acquisition().expect("second acquisition").usage(),
                 ConnectionUsage::Reused
             );
             let first_connection = first.connection().expect("first connection metadata");
             let second_connection = second.connection().expect("second connection metadata");
             assert_eq!(
                 first_connection.connection_id(),
                 second_connection.connection_id()
             );
             assert_eq!(
                 first_connection.establishment(),
                 second_connection.establishment(),
                 "reuse must retain the establishment that created the connection"
             );
             assert!(first_connection.establishment().is_some());
         } else {
             assert!(first.acquisition().is_none());
             assert!(first.connection().is_none());
             assert!(second.acquisition().is_none());
             assert!(second.connection().is_none());
-+
+        }
-+
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-+
+    }
-+
     #[tokio::test]
     async fn test_attempt_telemetry_with_hyper_util_legacy_pool() {
         attempt_telemetry_matches_backend_capability(&HyperUtilLegacyPool, false).await;
-+
+    }
-+
     #[tokio::test]
     async fn test_attempt_telemetry_with_partitioned_pool() {
         attempt_telemetry_matches_backend_capability(&PartitionedConnectionPool, true).await;
-+
+    }
-+
     #[tokio::test]
     async fn h2_acquisition_ends_before_response_headers() {
         let response_gate = ManualGate::new();
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([H2ConnectionScript::new()
                 .fallback(H2StreamScript::respond_after(
                     H2Response::ok("ok"),
                     response_gate.waiter(),
                 ))]))
             .start()
             .await
             .expect("H2 server should start");
         let time = ManualTimeSource::new(UNIX_EPOCH);
         let pool = ConnectionPool::builder()
             .tls_provider(rustls_aws_lc())
             .tls_context(test_tls::SERVER_IDENTITY.client_context())
             .time_source(time.clone())
             .build_https()
             .expect("valid partitioned HTTPS pool");
         let client =
             SharedHttpClient::new(PoolClient::new(&pool).expect("anonymous partition exists"));
         let connector = test_client::connector_with_time_source(&client, time.clone());
         let capture = CaptureHttpAttemptTelemetry::new();
         let mut request = HttpRequest::get(server.url("/gated")).expect("valid HTTP request");
         request.add_extension(capture.clone());
         let send = tokio::spawn({
             let connector = connector.clone();
             async move { test_client::send_and_collect(&connector, request).await }
         });
-+
         response_gate
             .wait_until_reached(test_client::WAIT)
             .await
             .expect("server should accept the H2 stream");
         time.advance(Duration::from_secs(10));
         response_gate.release();
         let (status, body) = send.await.expect("request task");
         assert_eq!((status, body.as_slice()), (200, b"ok".as_slice()));
-+
         let telemetry = capture.get();
         assert_eq!(
             telemetry.acquisition().expect("acquisition").duration(),
             Some(Duration::ZERO)
         );
         assert_eq!(
             telemetry.connector_call_duration(),
             Some(Duration::from_secs(10))
         );
-+
         drop(connector);
         drop(client);
         drop(pool);
         server.shutdown().await.expect("clean H2 server shutdown");
-+
+    }
-414
     /// Poisoning captured H2 connection metadata moves later streams to a new connection.
     async fn poisoned_connection_is_not_reused(backend: &dyn HttpsClientBackend) {
         let script =
             H2ConnectionScript::new().fallback(H2StreamScript::respond(H2Response::ok("ok")));
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([script.clone(), script]))
             .start()
             .await
             .expect("H2 server should start");

↧ Expand context

         let client = h2_client(backend);
         let connector = test_client::connector(&client);
-426
         let (status, body, metadata) =
             get_and_collect_with_capture(&connector, &server.url("/first")).await;
         assert_eq!((status, body.as_slice()), (200, b"ok".as_slice()));
         metadata.poison();
-431
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/second")).await;
         assert_eq!((status, body.as_slice()), (200, b"ok".as_slice()));
-434
         let first_connection = single_stream_connection(&server, "/first");
         let second_connection = single_stream_connection(&server, "/second");
         assert_ne!(first_connection, second_connection);
         assert_eq!(server.connection_count(), 2);
-439
         drop(metadata);
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");

@@ -612,748 +671,1050 @@

↥ Expand context

             .await
             .expect("H2 server should start");
         let client = h2_client(&PartitionedConnectionPool);
         let connector = test_client::connector(&client);
-752
         let outcome = tokio::time::timeout(
             test_client::WAIT,
             test_client::send_request(
                 &connector,
                 HttpRequest::get(server.url("/before-first-stream")).expect("valid HTTP request"),
             ),
-759
+        )
         .await
         .expect("GOAWAY before dispatch did not terminate");
         assert!(
             outcome.is_err(),
             "a request succeeded after GOAWAY excluded its stream"
         );
         assert!(
             server.connection_count() <= 3,

             "GOAWAY before dispatch created {} connections",
             server.connection_count()
         );
-771
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-775
+    }
-776
+}
-777
 mod abrupt_transport_failure {
     use super::*;
     use aws_smithy_http_client::pool::ConnectionEvent;
     use aws_smithy_runtime_api::client::connection::ConnectionId;
     use std::sync::{Arc, Mutex};
     use std::time::Duration;
-+
     #[derive(Clone, Copy, Debug, Eq, PartialEq)]
     enum PoolEvent {
         Opened(ConnectionId),
         LogicalClose(ConnectionId),
         PhysicalClose(ConnectionId),
-+
+    }
-+
     /// Raw transport loss terminates every accepted stream and later demand uses
     /// a replacement connection.
     async fn abrupt_transport_failure_terminates_streams_and_replaces_connection(
         client: SharedHttpClient,
     ) -> aws_smithy_http_client::pool::OriginKey {
         let body_gate = ManualGate::new();
         let held = |body| {
             H2StreamScript::respond(
                 H2Response::new(http_1x::StatusCode::OK).body(H2BodyPlan::gated(
                     body,
                     body_gate.waiter(),
                     "-never",
                 )),
-+
+            )
         };
         let first = H2ConnectionScript::new()
             .route("/warm", H2StreamScript::respond(H2Response::ok("warm")))
             .route("/one", held("one"))
             .route("/two", held("two"))
             .route("/three", held("three"));
         let second = H2ConnectionScript::new().route(
             "/replacement",
             H2StreamScript::respond(H2Response::ok("replacement")),
         );
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([first, second]))
             .start()
             .await
             .expect("H2 server should start");
         let connector = test_client::connector(&client);
-+
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/warm")).await;
         assert_eq!((status, body.as_slice()), (200, b"warm".as_slice()));
         let (one, two, three) = tokio::join!(
             test_client::send_request(
                 &connector,
                 HttpRequest::get(server.url("/one")).expect("valid HTTP request"),
             ),
             test_client::send_request(
                 &connector,
                 HttpRequest::get(server.url("/two")).expect("valid HTTP request"),
             ),
             test_client::send_request(
                 &connector,
                 HttpRequest::get(server.url("/three")).expect("valid HTTP request"),
             ),
         );
         let responses = [
             one.expect("first response headers should arrive"),
             two.expect("second response headers should arrive"),
             three.expect("third response headers should arrive"),
         ];
         body_gate
             .wait_for_arrivals(3, test_client::WAIT)
             .await
             .expect("all response streams should reach the body gate");
-+
         let original = single_stream_connection(&server, "/warm");
         assert_eq!(original, single_stream_connection(&server, "/one"));
         assert_eq!(original, single_stream_connection(&server, "/two"));
         assert_eq!(original, single_stream_connection(&server, "/three"));
         server
             .abort_transport(original)
             .await
             .expect("the original transport should abort");
-+
         for response in responses {
             let body_result =
                 tokio::time::timeout(test_client::WAIT, response.into_body().collect())
                     .await
                     .expect("aborted response body should terminate");
             assert!(
                 body_result.is_err(),
                 "abrupt transport loss completed an accepted response body"
             );
-+
+        }
         server
             .wait_for_event(test_client::WAIT, |event| {
                 matches!(
                     event,
                     H2Event::ConnectionClosed {
                         connection_id,
                         reason: ConnectionCloseReason::ScriptedTransportAbort,
                     } if *connection_id == original
-+
+                )
             })
             .await
             .expect("the harness should record raw transport loss");
-+
         let replacement_url = server.url("/replacement");
         let origin = aws_smithy_http_client::pool::OriginKey::from_uri(
             &replacement_url.parse().expect("valid replacement URI"),
-+
+        )
         .expect("replacement URI should name an origin");
         let (status, body) = test_client::get_and_collect(&connector, &replacement_url).await;
         assert_eq!((status, body.as_slice()), (200, b"replacement".as_slice()));
         assert_ne!(
             original,
             single_stream_connection(&server, "/replacement"),
             "later demand reused the aborted connection"
         );
         assert_eq!(2, server.connection_count());
-+
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
         origin
-+
+    }
-+
     async fn wait_for_pool_events(
         events: &Arc<Mutex<Vec<PoolEvent>>>,
         predicate: impl Fn(&[PoolEvent]) -> bool,
     ) {
         tokio::time::timeout(test_client::WAIT, async {
             loop {
                 if predicate(&events.lock().unwrap()) {
                     return;
-+
+                }
                 tokio::time::sleep(Duration::from_millis(10)).await;
-+
+            }
         })
         .await
         .expect("pool lifecycle events should converge");
-+
+    }
-+
     #[tokio::test]
     async fn test_abrupt_transport_failure_terminates_streams_and_replaces_connection_with_hyper_util_legacy_pool(
     ) {
         abrupt_transport_failure_terminates_streams_and_replaces_connection(h2_client(
             &HyperUtilLegacyPool,
         ))
         .await;
-+
+    }
-+
     #[tokio::test]
     async fn test_abrupt_transport_failure_terminates_streams_and_replaces_connection_with_partitioned_pool(
     ) {
         let events = Arc::new(Mutex::new(Vec::new()));
         let observed = events.clone();
         let pool = ConnectionPool::builder()
             .tls_provider(rustls_aws_lc())
             .tls_context(test_tls::SERVER_IDENTITY.client_context())
             .max_connections_per_host(1)
             .event_listener(move |event: &ConnectionEvent<'_>| {
                 let event = match event {
                     ConnectionEvent::Opened(opened) => {
                         Some(PoolEvent::Opened(opened.connection().id()))
-+
+                    }
                     ConnectionEvent::LogicalClose(closed) => {
                         Some(PoolEvent::LogicalClose(closed.connection().id()))
-+
+                    }
                     ConnectionEvent::PhysicalClose(closed) => {
                         Some(PoolEvent::PhysicalClose(closed.connection().id()))
-+
+                    }
                     ConnectionEvent::EstablishmentFailed(_) => None,
                     _ => None,
                 };
                 if let Some(event) = event {
                     observed.lock().unwrap().push(event);
-+
+                }
             })
             .build_https()
             .expect("valid partitioned HTTPS pool");
         let client = SharedHttpClient::new(
             PoolClient::new(&pool).expect("anonymous partition should resolve"),
         );
-+
         let origin =
             abrupt_transport_failure_terminates_streams_and_replaces_connection(client).await;
         wait_for_pool_events(&events, |events| {
             events
                 .iter()
                 .filter(|event| matches!(event, PoolEvent::PhysicalClose(_)))
                 .count()
                 >= 1
         })
         .await;
-+
         let events = events.lock().unwrap();
         let opened: Vec<_> = events
             .iter()
             .filter_map(|event| match event {
                 PoolEvent::Opened(connection) => Some(*connection),
                 _ => None,
             })
             .collect();
         assert_eq!(2, opened.len(), "expected original and replacement opens");
         assert_ne!(opened[0], opened[1]);
         assert_eq!(
 ,
             events
                 .iter()
                 .filter(|event| matches!(event, PoolEvent::LogicalClose(id) if *id == opened[0]))
                 .count(),
             "the aborted generation should close logically once"
         );
         assert_eq!(
 ,
             events
                 .iter()
                 .filter(|event| matches!(event, PoolEvent::PhysicalClose(id) if *id == opened[0]))
                 .count(),
             "the aborted transport should close physically once"
         );
         drop(events);
-+
         let stats = pool
             .partition_stats(PartitionId::ANONYMOUS, &origin)
             .expect("anonymous partition should retain origin statistics");
         assert_eq!(0, stats.pending_acquisitions());
         assert_eq!(0, stats.establishing_connections());
         assert_eq!(0, stats.h2().active_requests());
         assert_eq!(1, stats.h2().accepting());
         assert_eq!(0, stats.h2().draining());
         assert_eq!(1, stats.physically_live_connections());
         let capacity = pool
             .origin_stats(&origin)
             .capacity()
             .copied()
             .expect("origin should have bounded capacity");
         assert_eq!(1, capacity.limit());
         assert_eq!(
 ,
             capacity.in_use(),
             "replacement connection should own the returned slot"
         );
-+
+    }
-+
+}
-+
 #[cfg(feature = "s2n-tls")]
 mod protocol_negotiation {
     use super::*;
-1024
     /// The s2n-tls provider negotiates `h2` with ALPN and reuses that connection.
     async fn s2n_negotiates_h2_and_reuses_connection(backend: &dyn HttpsClientBackend) {
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([H2ConnectionScript::new()
                 .fallback(H2StreamScript::respond(H2Response::ok("s2n-h2")))]))
             .start()

↧ Expand context

             .await
             .expect("H2 server should start");
         let client = h2_client_with_provider(backend, tls::Provider::S2nTls);
         let connector = test_client::connector(&client);
-1035
         for request_number in 1..=3 {
             let (status, body) = test_client::get_and_collect(&connector, &server.url("/")).await;
             assert_eq!(status, 200, "request {request_number}");
             assert_eq!(body, b"s2n-h2", "request {request_number}");
-1040
+        }
-1041
         let negotiated = server
             .events()
             .into_iter()
             .filter_map(|event| match event {
                 H2Event::TlsNegotiated { alpn, .. } => Some(alpn),
                 _ => None,
             })
             .collect::<Vec<_>>();
         assert_eq!(negotiated, vec![Some(b"h2".to_vec())]);

@@ -1052,1431 +0,1697 @@

↥ Expand context

         assert_eq!((status, body.as_slice()), (200, b"partition".as_slice()));
         let first_connection = single_stream_connection(&server, "/first");
-1433
         let second_url = server.url("/second");
         let mut pending =
             tokio::spawn(async move { test_client::get_and_collect(&second, &second_url).await });
         assert!(
             tokio::time::timeout(Duration::from_millis(100), &mut pending)
                 .await
                 .is_err(),
             "an out-of-scope generation must not satisfy partition-local demand"
         );
         assert_eq!(server.connection_count(), 1);
-1444
         metadata.poison();
         let (status, body) = tokio::time::timeout(test_client::WAIT, pending)
             .await
             .expect("released capacity should wake partition-local demand")
             .expect("request task should not panic");
         assert_eq!((status, body.as_slice()), (200, b"partition".as_slice()));

         let second_connection = single_stream_connection(&server, "/second");
         assert_ne!(first_connection, second_connection);
         assert_eq!(server.connection_count(), 2);
-1454
         drop(metadata);
         drop(first);
         drop(pool);
         server.shutdown().await.expect("clean H2 server shutdown");
-1459
+    }
-1460
+}
-+
 mod runtime_placement {
     use super::*;
     use aws_smithy_http_client::pool::ConnectionEvent;
     use aws_smithy_runtime_api::client::connection::ConnectionId;
     use runtime::DrivenRuntime;
     use std::sync::{Arc, Mutex};
     use std::time::Duration;
     use tokio::runtime::{Handle, Id};
-+
     #[derive(Clone, Copy, Debug, Eq, PartialEq)]
     enum EventKind {
         Opened,
         LogicalClose,
         PhysicalClose,
-+
+    }
-+
     #[derive(Clone, Copy, Debug, Eq, PartialEq)]
     struct ObservedEvent {
         kind: EventKind,
         connection: ConnectionId,
         partition: PartitionId,
         runtime: Option<Id>,
-+
+    }
-+
     async fn wait_for_event(
         events: &Arc<Mutex<Vec<ObservedEvent>>>,
         predicate: impl Fn(&ObservedEvent) -> bool,
     ) -> ObservedEvent {
         tokio::time::timeout(test_client::WAIT, async {
             loop {
                 if let Some(event) = events.lock().unwrap().iter().copied().find(&predicate) {
                     return event;
-+
+                }
                 tokio::time::sleep(Duration::from_millis(10)).await;
-+
+            }
         })
         .await
         .expect("connection event should arrive")
-+
+    }
-+
     #[tokio::test]
     async fn peer_h2_reuse_keeps_connection_work_on_the_owner_runtime() {
         let first_id = PartitionId::from_index(1);
         let second_id = PartitionId::from_index(2);
         let first_runtime = DrivenRuntime::start("h2-owner-one");
         let first_runtime_id = first_runtime.id();
         let second_runtime = DrivenRuntime::start("h2-owner-two");
         let second_runtime_id = second_runtime.id();
         let peer_body_gate = ManualGate::new();
         let first_script = H2ConnectionScript::new()
             .abort_streams_on_client_close()
             .route("/first", H2StreamScript::respond(H2Response::ok("first")))
             .route(
                 "/peer",
                 H2StreamScript::respond(
                     H2Response::new(http_1x::StatusCode::OK).body(H2BodyPlan::gated(
                         "peer-",
                         peer_body_gate.waiter(),
                         "never",
                     )),
                 ),
             );
         let second_script = H2ConnectionScript::new().route(
             "/replacement",
             H2StreamScript::respond(H2Response::ok("replacement")),
         );
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([first_script, second_script]))
             .start()
             .await
             .expect("H2 server should start");
-+
         let events = Arc::new(Mutex::new(Vec::new()));
         let observed = events.clone();
         let pool = ConnectionPool::builder()
             .tls_provider(rustls_aws_lc())
             .tls_context(test_tls::SERVER_IDENTITY.client_context())
             .partitions([
                 Partition::new(first_id, first_runtime.driver_spawner()),
                 Partition::new(second_id, second_runtime.driver_spawner()),
             ])
             .connection_reuse_scope(ConnectionReuseScope::Pool)
             .max_connections_per_host(1)
             .event_listener(move |event: &ConnectionEvent<'_>| {
                 let observed_event = match event {
                     ConnectionEvent::Opened(opened) => {
                         Some((EventKind::Opened, opened.connection()))
-+
+                    }
                     ConnectionEvent::LogicalClose(closed) => {
                         Some((EventKind::LogicalClose, closed.connection()))
-+
+                    }
                     ConnectionEvent::PhysicalClose(closed) => {
                         Some((EventKind::PhysicalClose, closed.connection()))
-+
+                    }
                     ConnectionEvent::EstablishmentFailed(_) => None,
                     _ => None,
                 };
                 if let Some((kind, connection)) = observed_event {
                     observed.lock().unwrap().push(ObservedEvent {
                         kind,
                         connection: connection.id(),
                         partition: connection.owner_partition(),
                         runtime: Handle::try_current().ok().map(|handle| handle.id()),
                     });
-+
+                }
             })
             .build_https()
             .expect("valid partitioned HTTPS pool");
         let first_client = SharedHttpClient::new(
             PoolClient::from_partition(&pool, first_id).expect("first partition should resolve"),
         );
         let second_client = SharedHttpClient::new(
             PoolClient::from_partition(&pool, second_id).expect("second partition should resolve"),
         );
         let first = test_client::connector(&first_client);
         let second = test_client::connector(&second_client);
-+
         let first_request = first.clone();
         let first_url = server.url("/first");
         let (status, body) = first_runtime
             .spawn(async move { test_client::get_and_collect(&first_request, &first_url).await })
             .await
             .expect("first partition request task should not panic");
         assert_eq!((status, body.as_slice()), (200, b"first".as_slice()));
         let original_server_connection = single_stream_connection(&server, "/first");
         let original_open = wait_for_event(&events, |event| {
             event.kind == EventKind::Opened && event.partition == first_id
         })
         .await;
         assert_eq!(Some(first_runtime_id), original_open.runtime);
         assert!(first_runtime.submitted_tasks() > 0);
-+
         let peer_request = second.clone();
         let peer_url = server.url("/peer");
         let peer_response = second_runtime
             .spawn(async move {
                 test_client::send_request(
                     &peer_request,
                     HttpRequest::get(peer_url).expect("valid HTTP request"),
-+
+                )
                 .await
             })
             .await
             .expect("peer request task should not panic")
             .expect("peer response headers should arrive");
         peer_body_gate
             .wait_until_reached(test_client::WAIT)
             .await
             .expect("peer response should reach its body gate");
         assert_eq!(
             original_server_connection,
             single_stream_connection(&server, "/peer"),
             "peer demand should use the first partition's generation"
         );
         assert!(
             !events
                 .lock()
                 .unwrap()
                 .iter()
                 .any(|event| { event.kind == EventKind::Opened && event.partition == second_id }),
             "peer reuse should not establish a requester-owned connection"
         );
-+
         first_runtime.shutdown();
         let body_result =
             tokio::time::timeout(test_client::WAIT, peer_response.into_body().collect())
                 .await
                 .expect("owner-runtime shutdown should terminate the accepted response");
         assert!(
             body_result.is_err(),
             "accepted response completed after its owner runtime stopped"
         );
         let logical_close = wait_for_event(&events, |event| {
             event.kind == EventKind::LogicalClose && event.connection == original_open.connection
         })
         .await;
         assert_eq!(first_id, logical_close.partition);
         let physical_close = wait_for_event(&events, |event| {
             event.kind == EventKind::PhysicalClose && event.connection == original_open.connection
         })
         .await;
         assert_eq!(first_id, physical_close.partition);
-+
         let replacement_request = second.clone();
         let replacement_url = server.url("/replacement");
         let (status, body) = second_runtime
             .spawn(async move {
                 test_client::get_and_collect(&replacement_request, &replacement_url).await
             })
             .await
             .expect("replacement request task should not panic");
         assert_eq!((status, body.as_slice()), (200, b"replacement".as_slice()));
         assert_ne!(
             original_server_connection,
             single_stream_connection(&server, "/replacement")
         );
         let replacement_open = wait_for_event(&events, |event| {
             event.kind == EventKind::Opened && event.partition == second_id
         })
         .await;
         assert_eq!(Some(second_runtime_id), replacement_open.runtime);
         assert_ne!(original_open.connection, replacement_open.connection);
         assert!(second_runtime.submitted_tasks() > 0);
-+
         let origin = aws_smithy_http_client::pool::OriginKey::from_uri(
             &server.url("/").parse().expect("valid server URI"),
-+
+        )
         .expect("server URI should name an origin");
         let first_stats = pool
             .partition_stats(first_id, &origin)
             .expect("first partition should retain origin statistics");
         assert_eq!(0, first_stats.h2().active_requests());
         assert_eq!(0, first_stats.physically_live_connections());
         let second_stats = pool
             .partition_stats(second_id, &origin)
             .expect("second partition should retain origin statistics");
         assert_eq!(1, second_stats.h2().accepting());
         assert_eq!(0, second_stats.h2().active_requests());
         assert_eq!(1, second_stats.physically_live_connections());
         let capacity = pool
             .origin_stats(&origin)
             .capacity()
             .copied()
             .expect("origin should have bounded capacity");
         assert_eq!(1, capacity.limit());
         assert_eq!(1, capacity.in_use());
-+
         drop(first);
         drop(second);
         drop(first_client);
         drop(second_client);
         drop(pool);
         second_runtime.shutdown();
         server.shutdown().await.expect("clean H2 server shutdown");
-+
+    }
-+
+}

Pages: 1 2 3 4

1 1	`/*`
2 2	`* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.`
3 3	`* SPDX-License-Identifier: Apache-2.0`