AWS SDK

rev. 39e8f0a4a4cd7fc94cae1e2f796e9c21897d9cb9 (ignoring whitespace)

Files changed:

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

@@ -39,39 +104,98 @@

     /// Pool-owned clock used to assign and evaluate idle deadlines.
     time_source: SharedTimeSource,
     /// Pool-owned sleeper used by the partition task.
     sleep: Option<SharedAsyncSleep>,
     /// Fast start-once gate kept off the per-request lock path.
     started: AtomicBool,
     /// Registered cells, task state, and wake revision.
     state: Mutex<MaintenanceState>,
-47
+}
-48
 impl PartitionMaintenance {
     /// Creates a scheduler from immutable pool maintenance policy.
     pub(super) fn new(config: MaintenanceConfig) -> Arc<Self> {
         Arc::new(Self {
             idle_timeout: config.idle_timeout,
             time_source: config.time_source,
             sleep: config.sleep,
             started: AtomicBool::new(false),
             state: Mutex::new(MaintenanceState::default()),
         })

-59
+    }
-60
     /// Returns the deadline for a sender becoming idle now.
     pub(super) fn idle_deadline(&self) -> Option<SystemTime> {
         self.idle_timeout
             .and_then(|timeout| self.time_source.now().checked_add(timeout))
-65
+    }
-66
     /// Registers a retained cell without waking an idle task.
     pub(super) fn register(&self, cell: &Arc<OriginCell>) {
         // Loom has no modeled Weak. Skip this production-only index to avoid a
         // model-only maintenance -> cell -> maintenance cycle.
         #[cfg(all(test, smithy_http_client_loom))]
         let _ = cell;
--
         #[cfg(not(all(test, smithy_http_client_loom)))]
         self.state
             .lock()
             .cells
             .insert(cell.id().clone(), Weak::from_arc(cell));
-73
+    }
-74
     /// Publishes newly idle work when it precedes the current wakeup.
     pub(super) fn notify_deadline(&self, deadline: Option<SystemTime>) {
         let Some(deadline) = deadline else {
             return;

↧ Expand context

         };
         let wake = {
             let mut state = self.state.lock();
             if state.shutdown
                 || state
                     .scheduled_deadline
                     .is_some_and(|scheduled| scheduled <= deadline)
-86
+            {
                 return;
-88
+            }
             state.scheduled_deadline = Some(deadline);
             state.signal()
         };
         if let Some(waker) = wake {
             waker.wake();
-94
+        }
-95
+    }
-96
     /// Starts this partition's maintenance task at most once.
     pub(super) fn start(this: &Arc<Self>, spawner: &dyn DriverSpawner) {

@@ -164,158 +223,225 @@

↥ Expand context

         let mut state = self.state.lock();
         if state.shutdown {
             return ScheduleResult::Shutdown;
-161
+        }
         if state.revision != observed {
             return ScheduleResult::Retry;
-164
+        }
         state.scheduled_deadline = deadline;
         ScheduleResult::Wait {
             revision: state.revision,
             deadline,
-169
+        }
-170
+    }
-171
     /// Polls until a deadline publication or shutdown changes `observed`.
     fn poll_revision(&self, observed: u64, cx: &Context<'_>) -> Poll<()> {
         let mut state = self.state.lock();
         if state.shutdown || state.revision != observed {
             return Poll::Ready(());
-177
+        }

         if state
             .waker
             .as_ref()
             .is_none_or(|registered| !registered.will_wake(cx.waker()))
-182
+        {
             state.waker = Some(cx.waker().clone());
-184
+        }
         Poll::Pending
-186
+    }
-187
     #[cfg(all(test, smithy_http_client_loom))]
     pub(super) fn clear_modeled_cells_for_test(&self) {
         // Loom has no modeled Weak, so its synchronization facade retains
         // registered cells strongly. Explicit teardown prevents that model-only
         // substitution from creating a maintenance -> cell -> maintenance cycle.
         self.state.lock().cells.clear();
-+
+    }
-+
     #[cfg(all(test, feature = "rt-tokio", not(smithy_http_client_loom)))]
     fn probe(&self) -> MaintenanceProbe {
         let state = self.state.lock();
         MaintenanceProbe {
             started: self.started.load(Ordering::Acquire),
             shutdown: state.shutdown,
             scheduled_deadline: state.scheduled_deadline,
-203
+        }
-204
+    }
-205
+}

↧ Expand context

-206
 /// Mutable scheduler state protected independently from every origin cell.
 #[derive(Debug, Default)]
 struct MaintenanceState {
     /// Retained partition cells indexed for replacement on duplicate publish.
     cells: HashMap<CellId, Weak<OriginCell>>,
     /// Monotonic change counter observed by the maintenance future.
     revision: u64,
     /// Task waiting for a new revision.
     waker: Option<Waker>,
     /// Deadline the current sleep represents, if any.
     scheduled_deadline: Option<SystemTime>,
     /// Whether pool drop requested task termination.
     shutdown: bool,
-220
+}
-221
 impl MaintenanceState {
     /// Advances the wake revision and detaches the current task waker.
     fn signal(&mut self) -> Option<Waker> {
         self.revision = self

@@ -360,362 +511,599 @@

↥ Expand context

     observed: u64,
     sleep: SharedAsyncSleep,
     duration: Duration,
 ) -> MaintenanceWake {
     let mut sleeping = Box::pin(sleep.sleep(duration));
     poll_fn(|cx| {
         if sleeping.as_mut().poll(cx).is_ready() {
             return Poll::Ready(MaintenanceWake::DeadlineElapsed);
-370
+        }
         scheduler
             .upgrade()
             .map_or(Poll::Ready(MaintenanceWake::Revision), |current| {
                 current
                     .poll_revision(observed, cx)
                     .map(|()| MaintenanceWake::Revision)
             })
     })
     .await
-380
+}
-381

 #[cfg(all(test, feature = "rt-tokio", not(smithy_http_client_loom)))]
 struct MaintenanceProbe {
     started: bool,
     shutdown: bool,
     scheduled_deadline: Option<SystemTime>,
-387
+}
-388
 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use std::sync::atomic::{AtomicUsize, Ordering as StdOrdering};
     use crate::client::pool::cell::h1::{H1CloseHandle, H1Sender};
     use crate::client::pool::connection::{CloseReason, ConnectionInfo, ConnectionState};
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::{EligibilityGroup, PartitionId};
     use crate::sync::AtomicUsize;
     use aws_smithy_async::test_util::ManualTimeSource;
     use aws_smithy_runtime_api::client::connection::ConnectionId;
     use http_1x::uri::Scheme;
     use std::sync::Arc as StdArc;
     use std::task::{Context, Wake, Waker};
     use std::time::UNIX_EPOCH;
-403
     struct WakeCounter(AtomicUsize);
-405
     impl Wake for WakeCounter {
         fn wake(self: StdArc<Self>) {
             self.0.fetch_add(1, StdOrdering::SeqCst);
             self.0.fetch_add(1, Ordering::SeqCst);
-409
+        }
-410
         fn wake_by_ref(self: &StdArc<Self>) {
             self.0.fetch_add(1, StdOrdering::SeqCst);
             self.0.fetch_add(1, Ordering::SeqCst);
-413
+        }
-414
+    }
-415
     /// Races deadline publication with registration of the maintenance waker.
     ///
     /// A task that observes the old revision must either retry immediately or
     /// receive exactly one wake after it parks.
     #[test]
     fn deadline_publication_wakes_a_registered_task() {
         loom::model(|| {
             let maintenance = PartitionMaintenance::new(MaintenanceConfig::default());
             let observed = maintenance
                 .begin_scan()
                 .expect("new maintenance scheduler was shut down");
             let counter = StdArc::new(WakeCounter(AtomicUsize::new(0)));
             let waker = Waker::from(counter.clone());
             let mut context = Context::from_waker(&waker);
             let publisher = maintenance.clone();
             let publish = loom::thread::spawn(move || {
                 publisher.notify_deadline(Some(UNIX_EPOCH + Duration::from_secs(1)));
             });
-434
             let before_join = maintenance.poll_revision(observed, &mut context);
             publish.join().unwrap();
             if before_join.is_pending() {
                 assert_eq!(1, counter.0.load(StdOrdering::SeqCst));
                 assert_eq!(1, counter.0.load(Ordering::SeqCst));
-439
+            }
             assert!(maintenance.poll_revision(observed, &mut context).is_ready());
         });
-442
+    }
-443
     /// Publishes an earlier maintenance deadline while a scan is being committed.
     ///
     /// The scan must retry or retain a revision that forces the next poll to
     /// observe the concurrent publication.
     #[test]
     fn deadline_published_during_a_scan_forces_retry_or_wake() {
         loom::model(|| {
             let maintenance = PartitionMaintenance::new(MaintenanceConfig::default());
             let elapsed = UNIX_EPOCH + Duration::from_secs(1);
             let later = UNIX_EPOCH + Duration::from_secs(2);
             maintenance.notify_deadline(Some(elapsed));
             let observed = maintenance
                 .begin_scan()
                 .expect("new maintenance scheduler was shut down");
             assert_eq!(None, maintenance.state.lock().scheduled_deadline);
-459
             let publisher = maintenance.clone();
             let publish = loom::thread::spawn(move || publisher.notify_deadline(Some(later)));
             let result = maintenance.schedule(observed, None);
             publish.join().unwrap();
-464
             match result {
                 ScheduleResult::Retry => {}
                 ScheduleResult::Wait { revision, deadline } => {
                     assert_eq!(None, deadline);
                     let state = maintenance.state.lock();
                     assert_ne!(revision, state.revision);
                     assert_eq!(Some(later), state.scheduled_deadline);
-472
+                }
                 ScheduleResult::Shutdown => panic!("maintenance unexpectedly shut down"),
-474
+            }
         });
-476
+    }
-477
     /// Races shutdown with registration of the maintenance waker.
     ///
     /// A task that parks before shutdown must be woken and all later polls must
     /// observe terminal shutdown.
     #[test]
     fn shutdown_wakes_a_registered_task() {
         loom::model(|| {
             let maintenance = PartitionMaintenance::new(MaintenanceConfig::default());
             let observed = maintenance
                 .begin_scan()
                 .expect("new maintenance scheduler was shut down");
             let counter = StdArc::new(WakeCounter(AtomicUsize::new(0)));
             let waker = Waker::from(counter.clone());
             let mut context = Context::from_waker(&waker);
             let shutdown = maintenance.clone();
             let stop = loom::thread::spawn(move || shutdown.shutdown());
-494
             let before_join = maintenance.poll_revision(observed, &mut context);
             stop.join().unwrap();
             if before_join.is_pending() {
                 assert_eq!(1, counter.0.load(StdOrdering::SeqCst));
                 assert_eq!(1, counter.0.load(Ordering::SeqCst));
-499
+            }
             assert!(maintenance.poll_revision(observed, &mut context).is_ready());
         });
-502
+    }
-+
     /// Races an expired idle HTTP/1 record with local sender selection.
     ///
     /// Selection either takes the sender before the scan observes it or expiry
     /// closes the exact record. The sender and record must have one final owner.
     #[test]
     fn idle_expiry_linearizes_against_h1_selection() {
         loom::model(|| {
             let maintenance = PartitionMaintenance::new(MaintenanceConfig {
                 idle_timeout: Some(Duration::ZERO),
                 time_source: SharedTimeSource::new(ManualTimeSource::new(UNIX_EPOCH)),
                 sleep: None,
             });
             let cell = Arc::new(OriginCell::new(
                 PartitionId::from_index(1),
                 OriginKey::from_parts(Scheme::HTTP, "example.com", None).unwrap(),
                 EligibilityGroup::Pool,
                 None,
                 Some(maintenance.clone()),
             ));
             maintenance.register(&cell);
             let (connection, physical) = ConnectionState::unbounded(ConnectionInfo::for_test(
                 ConnectionId::new(1),
                 cell.id().partition(),
             ));
             OriginCell::insert_idle_h1(&cell, connection.clone(), H1Sender::test(1));
-+
             let selecting_cell = cell.clone();
             let selecting = loom::thread::spawn(move || OriginCell::select_h1(&selecting_cell));
             let scanning = maintenance.clone();
             let expiring = loom::thread::spawn(move || {
                 let observed = scanning
                     .begin_scan()
                     .expect("maintenance shut down before its scan");
                 let cells = scanning.cells();
                 for cell in &cells {
                     OriginCell::expire_idle(cell, UNIX_EPOCH);
-+
+                }
                 let nearest = cells
                     .iter()
                     .filter_map(|cell| cell.nearest_idle_deadline())
                     .min();
                 drop(cells);
                 scanning.schedule(observed, nearest)
             });
-+
             let selected = selecting.join().unwrap();
             assert!(matches!(
                 expiring.join().unwrap(),
                 ScheduleResult::Wait { .. }
             ));
             drop(selected);
-+
             if connection.probe().close_reason.is_none() {
                 assert!(H1CloseHandle::new(&cell, &connection).close(CloseReason::PoolDropped));
-+
+            }
             let stats = cell.connection_stats_snapshot();
             assert_eq!(0, stats.h1().idle());
             assert_eq!(0, stats.h1().active());
             assert!(matches!(
                 connection.probe().close_reason,
                 Some(CloseReason::IdleTimeout | CloseReason::PoolDropped)
             ));
             drop(physical);
             maintenance.clear_modeled_cells_for_test();
         });
-+
+    }
-570
+}
-571
 #[cfg(all(test, feature = "rt-tokio", not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
     use crate::client::pool::cell::h1::H1Sender;
     use crate::client::pool::connection::{ConnectionInfo, ConnectionProtocol, ConnectionState};
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::{EligibilityGroup, PartitionId, TokioDriverSpawner};
     use aws_smithy_async::test_util::controlled_time_and_sleep;

↧ Expand context

     use aws_smithy_async::time::TimeSource;
     use aws_smithy_runtime_api::client::connection::ConnectionId;
     use http_1x::uri::Scheme;
     use std::sync::atomic::{AtomicU64, Ordering};
     use std::time::{Duration, UNIX_EPOCH};
-585
     #[derive(Debug)]
     struct DroppingSpawner {
         submitted: Arc<AtomicU64>,
-589
+    }
-590
     impl DriverSpawner for DroppingSpawner {
         fn spawn(&self, driver: std::pin::Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             self.submitted.fetch_add(1, Ordering::SeqCst);
             drop(driver);
-595
+        }
-596
+    }
-597
     #[test]
     fn discarded_task_releases_the_start_latch() {

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

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

↥ Expand context

             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

@@ -1,1 +38,40 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 use crate::client::ConnectorCallTimer;
 use crate::hyper_legacy::timeout_middleware::HttpTimeoutError;
 use aws_smithy_async::future::timeout::TimedOutError;
 use aws_smithy_async::rt::sleep::{default_async_sleep, AsyncSleep, SharedAsyncSleep};
 use aws_smithy_async::time::SharedTimeSource;
 use aws_smithy_runtime_api::box_error::BoxError;
 use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
 use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
 use aws_smithy_runtime_api::client::connector_metadata::ConnectorMetadata;
 use aws_smithy_runtime_api::client::http::{
     HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpClient,
     SharedHttpConnector,
 };
 use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
 use aws_smithy_runtime_api::client::result::ConnectorError;

↧ Expand context

 use aws_smithy_runtime_api::client::runtime_components::{
     RuntimeComponents, RuntimeComponentsBuilder,
 };
 use aws_smithy_runtime_api::shared::IntoShared;
 use aws_smithy_types::body::SdkBody;
 use aws_smithy_types::config_bag::ConfigBag;
 use aws_smithy_types::error::display::DisplayErrorContext;
 use aws_smithy_types::retry::ErrorKind;
 use h2_0_3::Reason;
 use hyper_0_14::client::connect::{capture_connection, CaptureConnection, Connection, HttpInfo};
 use std::borrow::Cow;
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt;
 use std::sync::RwLock;
 use std::time::Duration;
 use tokio::io::{AsyncRead, AsyncWrite};
-38
 #[cfg(feature = "legacy-rustls-ring")]
 mod default_connector {

@@ -163,165 +405,418 @@

↥ Expand context

 /// [`HttpConnector`] that uses [`hyper_0_14`] to make HTTP requests.
 ///
 /// This connector also implements socket connect and read timeouts.
 ///
 /// This shouldn't be used directly in most cases.
 /// See the docs on [`HyperClientBuilder`] for examples of how
 /// to customize the Hyper client.
 #[derive(Debug)]
 pub struct HyperConnector {
     adapter: Box<dyn HttpConnector>,
-175
+}
-176
 impl HyperConnector {
     /// Builder for a Hyper connector.
     pub fn builder() -> HyperConnectorBuilder {
         Default::default()
-181
+    }
-182
+}
-183
 impl HttpConnector for HyperConnector {

     fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
         self.adapter.call(request)
-187
+    }
-188
+}
-189
 /// Builder for [`HyperConnector`].
 #[derive(Default, Debug)]
 pub struct HyperConnectorBuilder {
     connector_settings: Option<HttpConnectorSettings>,
     sleep_impl: Option<SharedAsyncSleep>,
     time_source: SharedTimeSource,
     client_builder: Option<hyper_0_14::client::Builder>,
-197
+}
-198
 impl HyperConnectorBuilder {
     /// Create a [`HyperConnector`] from this builder and a given connector.
     pub fn build<C>(self, tcp_connector: C) -> HyperConnector
     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>,
-208
+    {
         let client_builder = self.client_builder.unwrap_or_default();
         let sleep_impl = self.sleep_impl.or_else(default_async_sleep);
         let (connect_timeout, read_timeout) = self
             .connector_settings
             .map(|c| (c.connect_timeout(), c.read_timeout()))
             .unwrap_or((None, None));
-215
         let connector = match connect_timeout {
             Some(duration) => timeout_middleware::ConnectTimeout::new(
                 tcp_connector,
                 sleep_impl
                     .clone()
                     .expect("a sleep impl must be provided in order to have a connect timeout"),
                 duration,
             ),
             None => timeout_middleware::ConnectTimeout::no_timeout(tcp_connector),
         };
         let base = client_builder.build(connector);
         let read_timeout = match read_timeout {
             Some(duration) => timeout_middleware::HttpReadTimeout::new(
                 base,
                 sleep_impl.expect("a sleep impl must be provided in order to have a read timeout"),
                 duration,
             ),
             None => timeout_middleware::HttpReadTimeout::no_timeout(base),
         };
         HyperConnector {
             adapter: Box::new(Adapter {
                 client: read_timeout,
                 time_source: self.time_source,
             }),
-240
+        }
-241
+    }
-242
     /// Create a [`HyperConnector`] with the default rustls HTTPS implementation.
     #[cfg(feature = "legacy-rustls-ring")]
     pub fn build_https(self) -> HyperConnector {
         self.build(default_connector::https())
-247
+    }
-248
     /// Set the async sleep implementation used for timeouts
     ///
     /// Calling this is only necessary for testing or to use something other than
     /// [`default_async_sleep`].
     pub fn sleep_impl(mut self, sleep_impl: impl AsyncSleep + 'static) -> Self {
         self.sleep_impl = Some(sleep_impl.into_shared());
         self
-256
+    }
-257
     /// Set the async sleep implementation used for timeouts
     ///
     /// Calling this is only necessary for testing or to use something other than
     /// [`default_async_sleep`].
     pub fn set_sleep_impl(&mut self, sleep_impl: Option<SharedAsyncSleep>) -> &mut Self {
         self.sleep_impl = sleep_impl;
         self
-265
+    }
-266
     /// Copies the runtime-component clock used for request telemetry.
     fn set_time_source(&mut self, time_source: SharedTimeSource) -> &mut Self {
         self.time_source = time_source;
         self
-+
+    }
-+
     /// Configure the HTTP settings for the `HyperAdapter`
     pub fn connector_settings(mut self, connector_settings: HttpConnectorSettings) -> Self {
         self.connector_settings = Some(connector_settings);
         self
-277
+    }
-278
     /// Configure the HTTP settings for the `HyperAdapter`
     pub fn set_connector_settings(
         &mut self,
         connector_settings: Option<HttpConnectorSettings>,
     ) -> &mut Self {
         self.connector_settings = connector_settings;
         self
-286
+    }
-287
     /// Override the Hyper client [`Builder`](hyper_0_14::client::Builder) used to construct this client.
     ///
     /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.
     pub fn hyper_builder(mut self, hyper_builder: hyper_0_14::client::Builder) -> Self {
         self.client_builder = Some(hyper_builder);
         self
-294
+    }
-295
     /// Override the Hyper client [`Builder`](hyper_0_14::client::Builder) used to construct this client.
     ///
     /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.
     pub fn set_hyper_builder(
         &mut self,
         hyper_builder: Option<hyper_0_14::client::Builder>,
     ) -> &mut Self {
         self.client_builder = hyper_builder;
         self
-305
+    }
-306
+}
-307
 /// Adapter from a [`hyper_0_14::Client`] to [`HttpConnector`].
 ///
 /// This adapter also enables TCP `CONNECT` and HTTP `READ` timeouts via [`HyperConnector::builder`].
 struct Adapter<C> {
     client: timeout_middleware::HttpReadTimeout<
         hyper_0_14::Client<timeout_middleware::ConnectTimeout<C>, SdkBody>,
     >,
     time_source: SharedTimeSource,
-316
+}
-317
 impl<C> fmt::Debug for Adapter<C> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Adapter")
             .field("client", &"** hyper client **")
             .finish()
-323
+    }
-324
+}
-325
 /// Extract a smithy connection from a hyper CaptureConnection
 fn extract_smithy_connection(capture_conn: &CaptureConnection) -> Option<ConnectionMetadata> {
     let capture_conn = capture_conn.clone();
     if let Some(conn) = capture_conn.clone().connection_metadata().as_ref() {
         let mut extensions = http_02x::Extensions::new();
         conn.get_extras(&mut extensions);
         let http_info = extensions.get::<HttpInfo>();
         let mut builder = ConnectionMetadata::builder()
             .proxied(conn.is_proxied())
             .poison_fn(move || match capture_conn.connection_metadata().as_ref() {
                 Some(conn) => conn.poison(),
                 None => tracing::trace!("no connection existed to poison"),
             });
-339
         builder
             .set_local_addr(http_info.map(|info| info.local_addr()))
             .set_remote_addr(http_info.map(|info| info.remote_addr()));
-343
         let smithy_connection = builder.build();
-345
         Some(smithy_connection)
     } else {
         None
-349
+    }
-350
+}
-351
 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>,
-359
+{
     fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
         use hyper_0_14::service::Service;
-362
         let connector_call_timer = ConnectorCallTimer::start(&request, &self.time_source);
         let mut request = match request.try_into_http02x() {
             Ok(request) => request,
             Err(err) => {
                 return HttpConnectorFuture::ready(Err(ConnectorError::other(err.into(), None)));
-368
+            }
         };
         let capture_connection = capture_connection(&mut request);
         if let Some(capture_smithy_connection) =
             request.extensions().get::<CaptureSmithyConnection>()
-373
+        {
             capture_smithy_connection
                 .set_connection_retriever(move || extract_smithy_connection(&capture_connection));
-376
+        }
         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)),
             let result = 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)),
             };
             if let Some(timer) = connector_call_timer {
                 timer.finish();
-387
+            }
             result
         })
-390
+    }
-391
+}
-392
 /// 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);
-398
+    }

↧ Expand context

     // 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,
     };
-410
     // otherwise, we have no idea!
     ConnectorError::other(err, None)
-413
+}
-414
 /// 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() {
         return ConnectorError::timeout(err.into());

@@ -464,477 +523,539 @@

↥ Expand context

             .field("connector_cache", &self.connector_cache)
             .field("client_builder", &self.client_builder)
             .finish()
-480
+    }
-481
+}
-482
 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>,
-491
+{
     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());
                 if let Some(time_source) = components.time_source() {
                     builder.set_time_source(time_source);
-+
+                }
-510
                 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);
-517
+                    }
-518
+                }
                 let connector = SharedHttpConnector::new(builder.build(tcp_connector));

↧ Expand context

                 cache.insert(key.clone(), connector);
-521
+            }
             connector = cache.get(&key).cloned();
-523
+        }
-524
         connector.expect("cache populated above")
-526
+    }
-527
     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(())

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`.

@@ -219,219 +278,292 @@

↥ Expand context

             builder = builder.proxy_config(proxy_config);
-220
+        }
         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"))
-227
+    }
-228
+}
-229
 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")
-236
+}
-237
 pub(crate) fn connector_with_settings(

     client: &SharedHttpClient,
     settings: HttpConnectorSettings,
 ) -> SharedHttpConnector {
     client.http_connector(&settings, &runtime_components())
-243
+}
-244
 pub(crate) fn connector(client: &SharedHttpClient) -> SharedHttpConnector {
     connector_with_settings(client, HttpConnectorSettings::builder().build())
-247
+}
-248
 /// 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")
-270
+}
-271
 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
-280
+}
-281
 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"),
-286
+    )
     .await
-288
+}
-289
 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

@@ -170,170 +292,322 @@

↥ Expand context

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

-190
     pub(crate) fn body(mut self, body: H2BodyPlan) -> Self {
         self.body = body;
         self
-194
+    }
-195
+}
-196
 #[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),
-211
+}
-212
 impl H2StreamScript {
     pub(crate) fn respond(response: H2Response) -> Self {
         Self::Respond(response)
-216
+    }
-217
     /// 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 }
-229
+    }
-230
     pub(crate) fn reset(reason: Reason) -> Self {
         Self::Reset(reason)
-233
+    }
-234
+}
-235
 #[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,
-253
+}
-254
 impl H2ConnectionScript {
     pub(crate) fn new() -> Self {
         Self::default()
-258
+    }
-259
     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
-267
+    }
-268
     pub(crate) fn fallback(mut self, script: H2StreamScript) -> Self {
         self.fallback = Some(script);
         self
-272
+    }
-273
     pub(crate) fn allow_handshake_abandonment(mut self) -> Self {
         self.allow_handshake_abandonment = true;
         self
-277
+    }
-278
     pub(crate) fn goaway_on_ready(mut self) -> Self {
         self.goaway_on_ready = true;
         self
-282
+    }
-283
     /// 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()
-295
+    }
-296
+}
-297
 #[derive(Clone, Debug)]
 pub(crate) enum H2ConnectionPlan {
     Queue(VecDeque<H2ConnectionScript>),
     Unbounded(H2ConnectionScript),
-302
+}

↧ Expand context

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

@@ -367,397 +426,457 @@

↥ Expand context

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

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

↧ Expand context

         self
-439
+    }
-440
     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(

@@ -506,537 +565,624 @@

↥ Expand context

-537
     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"
                 ))
             })
-565
+    }
-566
     /// 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}"
                 )))
-607
+            }
             Err(_) => {
                 listener_task.abort();
                 return Err(H2HarnessError::new(
                     "timed out waiting for H2 listener and connection tasks to stop",
                 ));
-613
+            }
-614
+        }
         if let Some(failure) = self.state.failure() {
             return Err(failure);
-617
+        }
         Ok(())
-619
+    }
-620
+}
-621
 impl Drop for H2TestServer {
     fn drop(&mut self) {
         self.shutdown.send_replace(true);

@@ -710,769 +790,855 @@

↥ 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:?}"
         )));
-776
+    }
-777
     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);

-789
+            }
             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);
-795
+    }
-796
     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;
-805
     loop {
         tokio::select! {
             changed = shutdown.changed(), if !shutting_down => {
                 if changed.is_err() || *shutdown.borrow() {
                     shutting_down = true;
                     connection.abrupt_shutdown(Reason::NO_ERROR);
-812
+                }
-813
+            }
             command = control.recv(), if control_open => {
                 match command {
                     Some(ConnectionCommand::GracefulShutdown(acknowledged)) => {
                         graceful_shutdown = true;
                         connection.graceful_shutdown();
                         let _ = acknowledged.send(());
-820
+                    }
                     Some(ConnectionCommand::AbortTransport(acknowledged)) => {
                         transport_aborted = true;
                         let _ = acknowledged.send(());
                         break;
-+
+                    }
                     None => control_open = false,
-827
+                }
-828
+            }
             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();
                         state.record_event(H2Event::StreamAccepted {

↧ Expand context

                             connection_id,
                             stream_id,
                             method,
                             path: path.clone(),
                         });
                         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()))
-850
+                            }
                             _ => {
                                 drop(request);
                                 None
-854
+                            }
                         };

@@ -802,867 +917,1019 @@

↥ Expand context

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

                         break;
-888
+                    }
-889
+                }
-890
+            }
             completed = stream_tasks.join_next(), if !stream_tasks.is_empty() => {
                 record_stream_task_result(&state, completed);
-893
+            }
-894
+        }
-895
+    }
-896
     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));
-913
+            }
         };
         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() {}
-921
+        }
         if client_initiated_close {
             Ok(ConnectionCloseReason::ClientClosed)
         } else {
             Ok(ConnectionCloseReason::ScriptCompleted)
-926
+        }
-927
+    }
-928
+}
-929
 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}")))
-962
+        }
-963
+    }
-964
+}
-965
 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,
             });
-987
+        }
         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(());
-1012
+            }
-1013
             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,133 @@

 /*
  * 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 checks task placement on this runtime.
     pub(crate) fn driver_spawner(&self) -> RuntimeDriverSpawner {
         RuntimeDriverSpawner {
             handle: self.handle.clone(),
             runtime_id: self.runtime_id,
             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 have started on 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,
     runtime_id: Id,
     submitted_tasks: Arc<AtomicUsize>,
-+
+}
-+
 impl DriverSpawner for RuntimeDriverSpawner {
     fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
         let runtime_id = self.runtime_id;
         let submitted_tasks = self.submitted_tasks.clone();
         drop(self.handle.spawn(async move {
             assert_eq!(
                 runtime_id,
                 Handle::current().id(),
                 "connection-owned task started on the wrong runtime"
             );
             submitted_tasks.fetch_add(1, Ordering::SeqCst);
             driver.await;
         }));
-+
+    }
-+
+}

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

@@ -1,1 +64,71 @@

↥ 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 common::client as test_client;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use std::time::{Duration, UNIX_EPOCH};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpListener;
-44
 const IP1: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
 const IP2: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2));
-47
 #[derive(Debug)]
 enum ObservedConnectionEvent {
     Failed {
         establishment: ConnectionEstablishmentId,

↧ Expand context

         origin: OriginKey,
         stage: ConnectionEstablishmentStage,
         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,
     },
-70
+}
-71

@@ -292,299 +407,582 @@

↥ Expand context

-299
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             Http1Script::responses([Http1Response::ok().body("custom dns")]),
-304
+        )
         .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());
-318

     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());
-322
     drop(connector);
     drop(client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-327
+}
-328
 #[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"),
             ]),
-500
+        )
         .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());
-511
     let first = test_client::send_request(
         &connector,
         HttpRequest::get(harness.endpoint_url()).expect("valid request"),
-515
+    )
     .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");
-522
     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());
-537
     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))
     );
-553
     drop(connector);
     drop(client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-558
+}
-559
 #[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"),
             ]),
-569
+        )
         .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,585 +469,734 @@

↥ Expand context

     let second_client = shared_client(&pool, second_id);
     let first_connector = connector(&first_client);
     let second_connector = connector(&second_client);
-588
     test_client::get_and_collect(&first_connector, &harness.endpoint_url()).await;
     test_client::get_and_collect(&second_connector, &harness.endpoint_url()).await;
-591
     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());
-606
     drop(first_connector);
     drop(second_connector);
     drop(first_client);
     drop(second_client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-613
+}
-614
 #[tokio::test]
 async fn peer_h1_reuse_survives_borrower_runtime_shutdown() {
     use runtime::DrivenRuntime;
-+
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             Http1Script::responses([
                 Http1Response::ok().body("owner"),
                 Http1Response::ok().body("peer"),
                 Http1Response::ok().body("reused"),
             ]),
-+
+        )
         .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 peer_request = second_connector.clone();
     let peer_url = url.clone();
     let (status, body) = second_runtime
         .spawn(async move { test_client::get_and_collect(&peer_request, &peer_url).await })
         .await
         .expect("peer request task should not panic");
     assert_eq!((status, body.as_slice()), (200, b"peer".as_slice()));
     assert!(first_runtime.submitted_tasks() > 0);
     assert!(second_runtime.submitted_tasks() > 0);
-+
     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!(3, 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")]),
             ]),
-714
+        )

↧ 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);
-733
     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`