AWS SDK

rev. 214b1cfffb5029555908aba78574468110a9ad23 (ignoring whitespace)

Files changed:

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

@@ -581,581 +640,652 @@

-581
+    }
-582
     fn managed_cell(
         timeout: Duration,
     ) -> (
         Arc<PartitionMaintenance>,
         Arc<OriginCell>,
         aws_smithy_async::test_util::SleepGate,
     ) {
         let (time, sleep, gate) = controlled_time_and_sleep(UNIX_EPOCH);
         let (maintenance, cell) = cell_with_maintenance(
             timeout,
             SharedTimeSource::new(time),
             SharedAsyncSleep::new(sleep),
         );
         (maintenance, cell, gate)
-597
+    }
-598
     fn connection(id: u64) -> Arc<ConnectionState> {
         let info = ConnectionInfo::new(

             ConnectionId::new(id),
             OriginKey::from_parts(Scheme::HTTP, "example.com", None).unwrap(),
             PartitionId::from_index(1),
             NegotiatedProtocol::Http1,
             hyper_util::client::legacy::connect::Connected::new(),
         );
         let (connection, _physical) = ConnectionState::unbounded(info);
         connection
-609
+    }
-610
     fn h2_connection(id: u64) -> Arc<ConnectionState> {
         let info = ConnectionInfo::new(
             ConnectionId::new(id),
             OriginKey::from_parts(Scheme::HTTP, "example.com", None).unwrap(),
             PartitionId::from_index(1),
             NegotiatedProtocol::Http2,
             hyper_util::client::legacy::connect::Connected::new(),
         );
         let (connection, _physical) = ConnectionState::unbounded(info);
         connection
-+
+    }
-+
     #[derive(Clone, Debug)]
     struct TrackingSpawner {
         submitted: Arc<AtomicU64>,
         active: Arc<AtomicU64>,
-627
+    }
-628
     impl TrackingSpawner {
         fn new() -> Self {
             Self {
                 submitted: Arc::new(AtomicU64::new(0)),

↧ Expand context

                 active: Arc::new(AtomicU64::new(0)),
-634
+            }
-635
+        }
-636
+    }
-637
     impl DriverSpawner for TrackingSpawner {
         fn spawn(&self, driver: std::pin::Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             self.submitted.fetch_add(1, Ordering::SeqCst);
             let active = self.active.clone();
             active.fetch_add(1, Ordering::SeqCst);
             tokio::spawn(async move {
                 struct ActiveGuard(Arc<AtomicU64>);
                 impl Drop for ActiveGuard {
                     fn drop(&mut self) {
                         self.0.fetch_sub(1, Ordering::SeqCst);
-648
+                    }
-649
+                }
                 let _guard = ActiveGuard(active);
                 driver.await;
             });

@@ -703,715 +762,827 @@

↥ Expand context

         assert_eq!(revision, maintenance.state.lock().revision);
         maintenance.notify_deadline(Some(earlier));
         assert!(maintenance.state.lock().revision > revision);
         assert_eq!(Some(earlier), maintenance.snapshot().scheduled_deadline);
-719
+    }
-720
     #[tokio::test]
     async fn idle_record_closes_at_its_fake_time_deadline() {
         let timeout = Duration::from_secs(10);
         let (maintenance, cell, mut gate) = managed_cell(timeout);
         let connection = connection(1);
         OriginCell::install_idle_h1(&cell, connection.clone(), H1Sender::test(1));
         PartitionMaintenance::start(&maintenance, &TokioDriverSpawner::current());
-728
         let sleep = gate.expect_sleep().await;
         assert_eq!(timeout, sleep.duration());
         assert_eq!(None, connection.snapshot().close_reason);
         sleep.allow_progress();
-733
         for _ in 0..10 {

             if connection.snapshot().close_reason == Some(super::super::CloseReason::IdleTimeout) {
                 break;
-737
+            }
             tokio::task::yield_now().await;
-739
+        }
         assert_eq!(
             Some(super::super::CloseReason::IdleTimeout),
             connection.snapshot().close_reason
         );
-744
+    }
     #[tokio::test]
     async fn h2_generation_closes_at_its_fake_time_deadline() {
         let timeout = Duration::from_secs(10);
         let (maintenance, cell, mut gate) = managed_cell(timeout);
         let connection = h2_connection(1);
         OriginCell::install_h2_for_test(&cell, connection.clone(), 1, maintenance.idle_deadline());
         PartitionMaintenance::start(&maintenance, &TokioDriverSpawner::current());
-+
         let sleep = gate.expect_sleep().await;
         assert_eq!(timeout, sleep.duration());
         assert_eq!(None, connection.snapshot().close_reason);
         sleep.allow_progress();
-+
         for _ in 0..10 {
             if connection.snapshot().close_reason == Some(super::super::CloseReason::IdleTimeout) {
                 break;
-+
+            }
             tokio::task::yield_now().await;
-+
+        }
         assert_eq!(
             Some(super::super::CloseReason::IdleTimeout),
             connection.snapshot().close_reason
         );
         assert_eq!(None, cell.accepting_h2_generation());
-+
+    }
-+
     #[test]
     fn accepted_h2_dispatch_resets_the_idle_deadline() {
         let timeout = Duration::from_secs(10);
         let time = RewindableTimeSource::new(0);
         let (_unused_time, sleep, _gate) = controlled_time_and_sleep(UNIX_EPOCH);
         let (maintenance, cell) = cell_with_maintenance(
             timeout,
             SharedTimeSource::new(time.clone()),
             SharedAsyncSleep::new(sleep),
         );
         let connection = h2_connection(1);
         OriginCell::install_h2_for_test(&cell, connection.clone(), 1, maintenance.idle_deadline());
         assert_eq!(Some(UNIX_EPOCH + timeout), cell.nearest_idle_deadline());
-+
         time.set(20);
         let mut activation =
             OriginCell::select_h2(&cell).expect("HTTP/2 generation was not selectable");
         let _dispatch_parts = activation.take_dispatch_parts();
         let dispatch = ConnectionState::try_commit_dispatch(&connection)
             .expect("HTTP/2 connection rejected dispatch");
         activation.accept(dispatch);
-+
         assert_eq!(
             Some(UNIX_EPOCH + Duration::from_secs(20) + timeout),
             cell.nearest_idle_deadline()
         );
-+
+    }
-798
     #[tokio::test]
     async fn selected_record_has_no_idle_deadline() {
         let timeout = Duration::from_secs(10);
         let (maintenance, cell, mut gate) = managed_cell(timeout);
         let connection = connection(1);
         OriginCell::install_idle_h1(&cell, connection.clone(), H1Sender::test(1));
         PartitionMaintenance::start(&maintenance, &TokioDriverSpawner::current());
-806
         let sleep = gate.expect_sleep().await;

↧ Expand context

         let selection = OriginCell::select_h1(&cell).expect("idle H1 was not selected");
         sleep.allow_progress();
         for _ in 0..10 {
             tokio::task::yield_now().await;
-812
+        }
-813
         assert_eq!(None, connection.snapshot().close_reason);
         drop(selection);
-816
+    }
-817
     #[tokio::test]
     async fn completed_sleep_is_an_expiration_floor_after_clock_moves_backward() {
         let timeout = Duration::from_secs(10);
         let clock = RewindableTimeSource::new(100);
         let (_unused_time, sleep, mut gate) = controlled_time_and_sleep(UNIX_EPOCH);
         let (maintenance, cell) = cell_with_maintenance(
             timeout,
             SharedTimeSource::new(clock.clone()),
             SharedAsyncSleep::new(sleep),
         );

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

@@ -78,78 +138,138 @@

↥ Expand context

         self
-79
+    }
-80
     /// Returns the configured network-interface name for validation.
     pub(super) fn interface_name(&self) -> Option<&str> {
         self.interface.as_deref()
-84
+    }
-85
     /// Returns this partition's declared identity.
     pub(super) fn id(&self) -> PartitionId {
         self.id
-89
+    }
-90
     /// Decomposes this declaration for immutable registry storage.
     pub(super) fn into_parts(self) -> (PartitionId, Arc<dyn DriverSpawner>, Option<Arc<str>>) {
         (self.id, self.spawner, self.interface)
-94
+    }
-95
+}
-96
 impl fmt::Debug for Partition {

     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Partition")
             .field("id", &self.id)
             .field("spawner", &self.spawner)
             .field("interface", &self.interface)
             .finish()
-104
+    }
-105
+}
-106
 /// Identifies one declared connection partition.
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub struct PartitionId(usize);
-110
 impl PartitionId {
     /// Reserved identity for the implicit default partition.
     pub const ANONYMOUS: Self = Self(usize::MAX);
-114
     /// Creates a partition identity from a caller-owned index.
     ///
     /// `usize::MAX` is reserved for [`Self::ANONYMOUS`] and is rejected when
     /// used in an explicit partition declaration.

↧ Expand context

     pub const fn from_index(index: usize) -> Self {
         Self(index)
-121
+    }
-122
     /// Returns whether this is the reserved anonymous partition identity.
     pub const fn is_anonymous(self) -> bool {
         self.0 == Self::ANONYMOUS.0
-126
+    }
-127
+}
-128
 /// Spawns connection-owned work on a partition's runtime.
 ///
 /// This includes protocol drivers and short-lived tasks that finish connection
 /// establishment, return, or maintenance without moving the underlying I/O to
 /// the requesting runtime.
 pub trait DriverSpawner: fmt::Debug + Send + Sync + 'static {
     /// Spawns connection-owned work on the partition's runtime.
     fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>);
-137
+}
-138

@@ -155,155 +216,221 @@

↥ Expand context

     pub fn current() -> Self {
         Self::from_handle(tokio::runtime::Handle::current())
-157
+    }
-158
     /// Uses a specific Tokio runtime handle.
     pub fn from_handle(handle: tokio::runtime::Handle) -> Self {
         Self { handle }
-162
+    }
-163
+}
-164
 #[cfg(feature = "rt-tokio")]
 impl DriverSpawner for TokioDriverSpawner {
     fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
         drop(self.handle.spawn(driver));
-169
+    }
-170
+}
-171
 /// Controls which partitions may dispatch through each other's connections.
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
 #[non_exhaustive]

 pub enum ConnectionReuseScope {
     /// A connection serves only requests from its owning partition.
     Partition,
     /// Partitions with the same interface binding may share connections.
     #[default]
     NetworkInterface,
     /// Every partition in the pool may share connections.
     Pool,
-183
+}
-184
 /// Identifies the exact set of partitions eligible to reuse a connection.
 #[derive(Clone, Debug, Eq, Hash, PartialEq)]
 /// Identifies the exact set of partitions allowed to share a connection.
 ///
 /// The configured [`ConnectionReuseScope`] maps each partition to one group:
 /// partition-local scope creates a group for one partition, network-interface
 /// scope groups equal interface bindings, and pool scope creates one group for
 /// the whole pool.
 #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub(crate) enum EligibilityGroup {
     /// Only the partition with this identity is eligible.
     Partition(PartitionId),
     /// Partitions with this exact interface binding are eligible.
     NetworkInterface(Option<Arc<str>>),
     /// Every partition in the pool is eligible.
     Pool,
-199
+}
-200
 #[cfg(test)]

↧ Expand context

 mod tests {
     use super::*;
     #[cfg(feature = "rt-tokio")]
     use std::sync::atomic::{AtomicBool, Ordering};
-206
     #[derive(Debug)]
     struct TestSpawner;
-209
     impl DriverSpawner for TestSpawner {
         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-213
+        }
-214
+    }
-215
     #[test]
     fn anonymous_identity_is_reserved() {
         assert!(PartitionId::ANONYMOUS.is_anonymous());
         assert!(PartitionId::from_index(usize::MAX).is_anonymous());
         assert!(!PartitionId::from_index(0).is_anonymous());
-221
+    }

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

@@ -10,10 +196,203 @@

↥ Expand context

 //! runtime placement and lazily retains one [`OriginCell`] per canonical
 //! origin.
-12
 use super::admission::OriginAdmission;
 use super::cell::OriginCell;
 use super::connection::CloseReason;
 use super::maintenance::{MaintenanceConfig, PartitionMaintenance};
 use super::origin::{InvalidOrigin, OriginKey, OriginLookup, SchemeKey};
 #[cfg(feature = "rt-tokio")]
 use super::partition::TokioDriverSpawner;
 use super::partition::{
     ConnectionReuseScope, DriverSpawner, EligibilityGroup, Partition, PartitionId,
 };
 use crate::sync::{Arc, Mutex, RwLock};
 use http_1x::Uri;
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt;
 use std::num::{NonZeroU16, NonZeroUsize};
 use std::sync::{Arc as StdArc, OnceLock};

-30
 /// Fixed partition set and the origin-wide admission states it shares.
 #[derive(Debug)]
 pub(crate) struct PartitionRegistry {
     /// Immutable partition identities resolved when a client is built.
     partitions: HashMap<PartitionId, Arc<PartitionState>>,
     /// Policy used to derive a new cell's eligibility group.
     reuse_scope: ConnectionReuseScope,
     /// Optional origin-wide bound used when admission is first created.
     max_connections_per_host: Option<NonZeroUsize>,
     /// Whether admission may close idle H2 capacity for H1-required demand.
     allow_h2_reclaim_for_h1: bool,
     /// Admission authorities retained by canonical origin.
     bounded_origins: Mutex<HashMap<OriginKey, Arc<OriginAdmission>>>,
-44
+}
-45
 impl PartitionRegistry {
     /// Creates the anonymous partition or validates and retains explicit partitions.
     pub(crate) fn new(
         partitions: Option<Vec<Partition>>,
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
         allow_h2_reclaim_for_h1: bool,
         maintenance: MaintenanceConfig,
     ) -> Result<Self, PartitionRegistryError> {
         let Some(partitions) = partitions else {
             let partition = Arc::new(PartitionState::anonymous(maintenance));
             let mut partitions = HashMap::new();
             partitions.insert(PartitionId::ANONYMOUS, partition);
             return Ok(Self {
                 partitions,
                 reuse_scope,
                 max_connections_per_host,
                 allow_h2_reclaim_for_h1,
                 bounded_origins: Mutex::new(HashMap::new()),
             });
         };
-67
         let mut by_id = HashMap::new();
         for partition in partitions {
             let id = partition.id();
             if id.is_anonymous() {
                 return Err(PartitionRegistryError::ReservedAnonymousPartition);
-73
+            }
             if by_id
                 .insert(
                     id,
                     Arc::new(PartitionState::explicit(partition, maintenance.clone())),
-78
+                )
                 .is_some()
-80
+            {
                 return Err(PartitionRegistryError::DuplicatePartition(id));
-82
+            }
-83
+        }
-84
         if by_id.is_empty() {
             return Err(PartitionRegistryError::EmptyExplicitPartitionSet);
-87
+        }
-88
         Ok(Self {
             partitions: by_id,
             reuse_scope,
             max_connections_per_host,
             allow_h2_reclaim_for_h1,
             bounded_origins: Mutex::new(HashMap::new()),
         })
-96
+    }
-97
     /// Resolves a retained partition once during client construction.
     pub(crate) fn partition(&self, id: PartitionId) -> Option<Arc<PartitionState>> {
         self.partitions.get(&id).cloned()
-101
+    }
-102
     /// Resolves the stable cell for an already-resolved partition and URI.
     pub(crate) fn resolve_cell(
         &self,
         partition: &PartitionState,
         uri: &Uri,
     ) -> Result<Arc<OriginCell>, InvalidOrigin> {
         let lookup = OriginLookup::from_uri(uri)?;
         if let Some(cell) = partition.find_cell(&lookup) {
             return Ok(cell);
-112
+        }
-113
         let eligibility_group = self.eligibility_group(partition);
         let origin = lookup.into_owned();
         let admission = self.origin_admission(&origin);
         let candidate = Arc::new(OriginCell::new(
             partition.id(),
             origin,
             eligibility_group,
             admission.clone(),
             Some(partition.maintenance.clone()),
         ));
         let cell = match admission {
             Some(admission) => OriginAdmission::register_cell(&admission, candidate),
             None => candidate,
         };
         let cell = partition.publish_cell(cell);
         partition.maintenance.register(&cell);
         Ok(cell)
-131
+    }
-132
     /// Returns the shared admission authority for a bounded origin.
     ///
     /// Unbounded registries return `None`. Concurrent first use of a bounded
     /// origin converges on one retained authority.
     fn origin_admission(&self, origin: &OriginKey) -> Option<Arc<OriginAdmission>> {
         let limit = self.max_connections_per_host?;
         let admission = {
             let mut origins = self.bounded_origins.lock();
             origins
                 .entry(origin.clone())
                 .or_insert_with(|| OriginAdmission::new(origin.clone(), limit))
                 .or_insert_with(|| {
                     OriginAdmission::new(origin.clone(), limit, self.allow_h2_reclaim_for_h1)
                 })
                 .clone()
         };
         Some(admission)
-149
+    }
-150
     /// Derives which partitions may relieve a new cell's bounded demand.
     fn eligibility_group(&self, partition: &PartitionState) -> EligibilityGroup {
         match self.reuse_scope {
             ConnectionReuseScope::Partition => EligibilityGroup::Partition(partition.id()),
             ConnectionReuseScope::NetworkInterface => {
                 EligibilityGroup::NetworkInterface(partition.interface().cloned())
-157
+            }
             ConnectionReuseScope::Pool => EligibilityGroup::Pool,
-159
+        }
-160
+    }
-161
     /// Stops partition maintenance and logically closes every connection.
     pub(crate) fn close_all(&self, reason: CloseReason) {
         for partition in self.partitions.values() {
             partition.shutdown_maintenance();
-166
+        }
         let cells = self
             .partitions
             .values()
             .flat_map(|partition| partition.cells())
             .collect::<Vec<_>>();
         for cell in cells {
             OriginCell::close_all_h1(&cell, reason);
             OriginCell::close_all(&cell, reason);
-174
+        }
-175
+    }
-176
+}
-177
 /// Runtime placement and retained origin cells for one partition.
 #[derive(Debug)]
 pub(crate) struct PartitionState {
     /// Stable identity copied into every cell and owned connection.
     id: PartitionId,
     /// Configured spawner, or the first spawner published for the anonymous

↧ Expand context

     /// partition.
     spawner: OnceLock<StdArc<dyn DriverSpawner>>,
     /// Network interface used for placement and reuse eligibility.
     interface: Option<StdArc<str>>,
     /// Cells retained for the lifetime of this partition.
     origins: RwLock<OriginMap>,
     /// Owner-runtime idle maintenance for this partition.
     maintenance: Arc<PartitionMaintenance>,
-192
+}
-193
 impl PartitionState {
     /// Creates the implicit partition whose spawner is published on first use.
     fn anonymous(maintenance: MaintenanceConfig) -> Self {
         Self {
             id: PartitionId::ANONYMOUS,
             spawner: OnceLock::new(),
             interface: None,
             origins: RwLock::new(OriginMap::default()),
             maintenance: PartitionMaintenance::new(maintenance),
-203
+        }

@@ -398,405 +471,480 @@

↥ Expand context

 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
     use crate::client::pool::admission::ProtocolRequirement;
     use std::future::Future;
     use std::pin::Pin;
     use std::sync::Barrier;
-412
     #[derive(Debug)]
     struct TestSpawner;
-415
     impl DriverSpawner for TestSpawner {
         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-419
+        }
-420
+    }
-421
     fn partition(index: usize) -> Partition {
         Partition::new(PartitionId::from_index(index), TestSpawner)
-424
+    }

-425
     #[cfg(feature = "rt-tokio")]
     fn anonymous_registry(
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
     ) -> PartitionRegistry {
         PartitionRegistry::new(
             None,
             reuse_scope,
             max_connections_per_host,
             true,
             MaintenanceConfig::default(),
-437
+        )
         .unwrap()
-439
+    }
-440
     fn explicit_registry(
         partitions: impl IntoIterator<Item = Partition>,
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
     ) -> Result<PartitionRegistry, PartitionRegistryError> {
         PartitionRegistry::new(
             Some(partitions.into_iter().collect()),
             reuse_scope,
             max_connections_per_host,
             true,
             MaintenanceConfig::default(),
-452
+        )
-453
+    }
-454
     fn explicit_partition(registry: &PartitionRegistry, index: usize) -> Arc<PartitionState> {
         registry
             .partition(PartitionId::from_index(index))
             .expect("test partition was not registered")
-459
+    }
-460

↧ Expand context

     #[test]
     #[cfg(feature = "rt-tokio")]
     fn anonymous_partition_publishes_one_spawner() {
         const THREADS: usize = 8;
-465
         let runtime = tokio::runtime::Builder::new_current_thread()
             .build()
             .unwrap();
         let handle = runtime.handle().clone();
         let registry = Arc::new(anonymous_registry(
             ConnectionReuseScope::NetworkInterface,
             None,
         ));
         let partition = registry.partition(PartitionId::ANONYMOUS).unwrap();
         let barrier = Arc::new(std::sync::Barrier::new(THREADS));
         let mut threads = Vec::new();
-477
         for _ in 0..THREADS {
             let handle = handle.clone();
             let partition = partition.clone();

@@ -695,704 +785,795 @@

↥ Expand context

                 ConnectionReuseScope::default(),
                 NonZeroUsize::new(1),
-706
+            )
             .unwrap(),
         );
         let partition = explicit_partition(&registry, 1);
         let barrier = Arc::new(Barrier::new(THREADS));
         let mut threads = Vec::new();
-712
         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()
             }));
-722
+        }
-723

         let cells: Vec<_> = threads
             .into_iter()
             .map(|thread| thread.join().unwrap())
             .collect();
         assert!(cells
             .iter()
             .skip(1)
             .all(|cell| Arc::ptr_eq(&cells[0], cell)));
         assert_eq!(1, partition.cell_count());
-733
         let waiter = cells[0].register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cells[0], ProtocolRequirement::H1Compatible);
         let lease = OriginCell::take_ready_lease(&cells[0], waiter)
             .expect("admission targeted a different cell than the registry retained");
         drop(lease);
-738
+    }
-739
+}
-740
 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use std::future::Future;
     use std::pin::Pin;
-746
     #[derive(Debug)]
     struct TestSpawner;
-749
     impl DriverSpawner for TestSpawner {
         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-753
+        }
-754
+    }
-755
     fn explicit_registry(
         partitions: impl IntoIterator<Item = Partition>,
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
     ) -> Result<PartitionRegistry, PartitionRegistryError> {
         PartitionRegistry::new(
             Some(partitions.into_iter().collect()),
             reuse_scope,
             max_connections_per_host,
             true,
             MaintenanceConfig::default(),
-767
+        )
-768
+    }
-769
     #[test]
     fn first_cell_publication_is_stable_under_contention() {
         loom::model(|| {
             let registry = Arc::new(
                 explicit_registry(
                     [Partition::new(PartitionId::from_index(1), TestSpawner)],

↧ Expand context

                     ConnectionReuseScope::default(),
                     None,
-778
+                )
                 .unwrap(),
             );
             let partition = registry.partition(PartitionId::from_index(1)).unwrap();
-782
             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()

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

@@ -282,282 +354,371 @@

↥ Expand context

-282
+                }
-283
+            }
-284
+        }
-285
         fn verify_tls12_signature(
             &self,
             message: &[u8],
             cert: &CertificateDer<'_>,
             dss: &rustls::DigitallySignedStruct,
         ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
             self.web_pki_server_verifier
                 .verify_tls12_signature(message, cert, dss)
-294
+        }
-295
         fn verify_tls13_signature(
             &self,
             message: &[u8],
             cert: &CertificateDer<'_>,
             dss: &rustls::DigitallySignedStruct,
         ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {

             self.web_pki_server_verifier
                 .verify_tls13_signature(message, cert, dss)
-304
+        }
-305
         fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
             self.web_pki_server_verifier.supported_verify_schemes()
-308
+        }
-309
+    }
-310
     pub(crate) fn wrap_connector<R>(
         conn: HttpConnector<R>,
         crypto_mode: CryptoMode,
         tls_context: &TlsContext,
         proxy_config: crate::client::proxy::ProxyConfig,
     ) -> super::connect::RustTlsConnector<R> {
         wrap_connector_with_alpn(
             conn,
             crypto_mode,
             tls_context,
             proxy_config,
             &[b"h2", b"http/1.1"],
-+
+        )
-+
+    }
-+
     /// Wraps an HTTP connector with the ALPN offer selected for one pool attempt.
     pub(crate) fn wrap_connector_with_alpn<R>(
         mut conn: HttpConnector<R>,
         crypto_mode: CryptoMode,
         tls_context: &TlsContext,
         proxy_config: crate::client::proxy::ProxyConfig,
         alpn_protocols: &[&[u8]],
     ) -> super::connect::RustTlsConnector<R> {
         let client_config = create_rustls_client_config(crypto_mode, tls_context);
         let mut client_config = create_rustls_client_config(crypto_mode, tls_context);
         client_config.alpn_protocols = alpn_protocols
             .iter()
             .map(|protocol| protocol.to_vec())
             .collect();
         conn.enforce_http(false);
         let https_connector = hyper_rustls::HttpsConnectorBuilder::new()
             .with_tls_config(client_config.clone())
             .https_or_http()
             .enable_http1()
             .enable_http2()
             .wrap_connector(conn);
         let https_connector =
             hyper_rustls::HttpsConnector::from((conn, client_config.clone()));
-342
         super::connect::RustTlsConnector::new(https_connector, client_config, proxy_config)
-344
+    }
-345
+}
-346
 pub(crate) mod connect {
     use crate::client::connect::{Conn, Connecting};
     use crate::client::proxy::ProxyConfig;
     use aws_smithy_runtime_api::box_error::BoxError;
     use http_1x::uri::Scheme;

↧ Expand context

     use http_1x::Uri;
     use hyper::rt::{Read, ReadBufCursor, Write};
     use hyper_rustls::MaybeHttpsStream;
     use hyper_util::client::legacy::connect::{Connected, Connection, HttpConnector};
     use hyper_util::client::proxy::matcher::Matcher;
     use hyper_util::rt::TokioIo;
     use pin_project_lite::pin_project;
     use std::error::Error;
     use std::sync::Arc;
     use std::{
         io::{self, IoSlice},
         pin::Pin,
         task::{Context, Poll},
     };
     use tokio::io::{AsyncRead, AsyncWrite};
     use tokio::net::TcpStream;
     use tokio_rustls::client::TlsStream;
     use tower::Service;
-370
     #[derive(Debug, Clone)]

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

@@ -1,1 +47,48 @@

↥ Expand context

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Scripted HTTP/2 server support for connection behavior tests.
 //!

 //! This fixture is intentionally test-local. The project keeps its TLS setup surface
 //! out of the public API while still testing HTTP/2 connection-pool behavior against
 //! a real TLS+h2 server.
-11
 use super::client::WAIT;
 use super::tls;
 use aws_smithy_http_client::test_util::wire::connection::{ConnectionCloseReason, GateWaiter};
 use bytes::Bytes;
 use h2::server::SendResponse;
 use h2::Reason;
 use h2::RecvStream;
 use http_1x::{Method, Response, StatusCode};
 use std::collections::{HashMap, VecDeque};
 use std::error::Error;
 use std::fmt;
 use std::future::poll_fn;
 use std::io;
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};

↧ Expand context

 use std::task::{Context, Poll};
 use std::time::Duration;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::{mpsc, oneshot, watch};
 use tokio::task::{JoinHandle, JoinSet};
 use tokio_rustls::TlsAcceptor;
-36
 const H2_FRAME_HEADER_LENGTH: usize = 9;
 const GOAWAY_FRAME_TYPE: u8 = 0x7;
 const GOAWAY_PAYLOAD_LENGTH: usize = 8;
 /// Mask for the 31-bit stream identifier in a GOAWAY payload; the high bit is reserved.
 const GOAWAY_LAST_STREAM_ID_MASK: u32 = 0x7fff_ffff;
-42
 /// Test-local error for the H2 fixture. Structurally identical to the public
 /// `HarnessError` but kept local because `HarnessError::new` is not public.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(crate) struct H2HarnessError {
     message: Arc<str>,
-48
+}

@@ -169,170 +272,292 @@

↥ 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),
     /// Sends the response before a gate permits the request body to be drained.
     RespondBeforeReceivingRequestBody {
         response: H2Response,
         receive: GateWaiter,
     },
     Reset(Reason),
-206
+}
-207
 impl H2StreamScript {
     pub(crate) fn respond(response: H2Response) -> Self {
         Self::Respond(response)
-211
+    }
-212
     /// 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 }
-+
+    }
-+
     pub(crate) fn reset(reason: Reason) -> Self {
         Self::Reset(reason)
-223
+    }
-224
+}
-225
 #[derive(Clone, Debug, Default)]
 pub(crate) struct H2ConnectionScript {
     routes: HashMap<String, H2StreamScript>,
     fallback: Option<H2StreamScript>,
     allow_handshake_abandonment: bool,
     goaway_on_ready: bool,
-232
+}
-233
 impl H2ConnectionScript {
     pub(crate) fn new() -> Self {
         Self::default()
-237
+    }
-238
     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
-246
+    }
-247
     pub(crate) fn fallback(mut self, script: H2StreamScript) -> Self {
         self.fallback = Some(script);
         self
-251
+    }
-252
     pub(crate) fn allow_handshake_abandonment(mut self) -> Self {
         self.allow_handshake_abandonment = true;
         self
-256
+    }
-257
     pub(crate) fn goaway_on_ready(mut self) -> Self {
         self.goaway_on_ready = true;
         self
-+
+    }
-+
     fn script_for(&self, path: &str) -> Option<H2StreamScript> {
         self.routes.get(path).or(self.fallback.as_ref()).cloned()
-265
+    }
-266
+}
-267
 #[derive(Clone, Debug)]
 pub(crate) enum H2ConnectionPlan {
     Queue(VecDeque<H2ConnectionScript>),
     Unbounded(H2ConnectionScript),
-272
+}

↧ Expand context

-273
 impl H2ConnectionPlan {
     pub(crate) fn queue(scripts: impl IntoIterator<Item = H2ConnectionScript>) -> Self {
         Self::Queue(scripts.into_iter().collect())
-277
+    }
-278
     pub(crate) fn unbounded(script: H2ConnectionScript) -> Self {
         Self::Unbounded(script)
-281
+    }
-282
     fn next_script(&mut self) -> Option<H2ConnectionScript> {
         match self {
             Self::Queue(scripts) => scripts.pop_front(),
             Self::Unbounded(script) => Some(script.clone()),
-287
+        }
-288
+    }
-289
+}
-290
 #[derive(Debug)]
 struct RecordedState {

@@ -683,703 +806,836 @@

↥ Expand context

     let alpn = tls_stream
         .get_ref()
         .1
         .alpn_protocol()
         .map(ToOwned::to_owned);
     state.record_event(H2Event::TlsNegotiated {
         connection_id,
         alpn: alpn.clone(),
     });
     if alpn.as_deref() != Some(b"h2") {
         return Err(H2HarnessError::new(format!(
             "connection {connection_id:?} negotiated unexpected ALPN {alpn:?}"
         )));
-716
+    }
-717
     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);
-729
+            }
             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);
-+
+    }
-736
     let mut stream_tasks = JoinSet::new();
     let mut shutting_down = false;
     let mut graceful_shutdown = 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;
-744
     loop {
         tokio::select! {
             changed = shutdown.changed(), if !shutting_down => {
                 if changed.is_err() || *shutdown.borrow() {
                     shutting_down = true;
                     connection.abrupt_shutdown(Reason::NO_ERROR);
-751
+                }
-752
+            }
             command = control.recv(), if control_open => {
                 match command {
                     Some(ConnectionCommand::GracefulShutdown(acknowledged)) => {
                         graceful_shutdown = true;
                         connection.graceful_shutdown();
                         let _ = acknowledged.send(());
-759
+                    }
                     None => control_open = false,
-761
+                }
-762
+            }
             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 {
                             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:?}"
                             )));
                         };
                         // Scripts cover the response side only, so the request body is
                         // never read. Dropping the `RecvStream` marks the stream as no
                         // longer receiving: h2 then discards incoming DATA without
                         // charging the stream window and releases the connection
                         // capacity, so a client streaming a request body runs to
                         // completion. Holding it instead would charge the window on
                         // every DATA frame and never refund it, stalling the client
                         // once the send window is exhausted.
                         //
                         // TODO(test-utils): to script request bodies, pass
                         // `request.into_body()` into `run_stream` and drive it there,
                         // calling `release_capacity()` as chunks are consumed.
                         let held_request = match &stream_script {
                             H2StreamScript::RespondBeforeReceivingRequestBody { receive, .. } => {
                                 Some((request.into_body(), receive.clone()))
-+
+                            }
                             _ => {
                                 drop(request);
                                 None
-+
+                            }
                         };
                         // Most scripts cover the response side only. Dropping their
                         // `RecvStream` marks the stream as no longer receiving: h2
                         // then discards incoming DATA without charging the stream
                         // window and releases connection capacity, so a client
                         // streaming a request body runs to completion.
                         //
                         // A lifetime contract may retain a pending request body
                         // behind a gate without reading DATA. General request-body
                         // scripts must drive `RecvStream` and call
                         // `release_capacity()` as chunks are consumed.
                         stream_tasks.spawn(run_stream(
                             connection_id,
                             stream_id,
                             stream_script,
                             respond,
                             state.clone(),
                             held_request,
                         ));
-808
+                    }
                     Some(Err(err)) => {
                         if shutting_down
                             || (graceful_shutdown && err.reason() == Some(Reason::NO_ERROR))
-812
+                        {
                             break;
-814
+                        }
                         return Err(H2HarnessError::new(format!(
                             "H2 connection {connection_id:?} failed while accepting a stream: {err}"

↧ Expand context

                         )));
-818
+                    }
                     None => {
                         client_initiated_close = true;
                         break;
-822
+                    }
-823
+                }
-824
+            }
             completed = stream_tasks.join_next(), if !stream_tasks.is_empty() => {
                 record_stream_task_result(&state, completed);
-827
+            }
-828
+        }
-829
+    }
-830
     if shutting_down {
         stream_tasks.abort_all();
         while stream_tasks.join_next().await.is_some() {}
         Ok(ConnectionCloseReason::HarnessShutdown)
     } else {
         let drain = async {

@@ -815,845 +901,937 @@

↥ Expand context

             stream_tasks.abort_all();
             while stream_tasks.join_next().await.is_some() {}
-847
+        }
         if client_initiated_close {
             Ok(ConnectionCloseReason::ClientClosed)
         } else {
             Ok(ConnectionCloseReason::ScriptCompleted)
-852
+        }
-853
+    }
-854
+}
-855
 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}")))

-865
+        }
-866
+    }
-867
+}
-868
 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> {
     match script {
         H2StreamScript::Reset(reason) => {
             respond.send_reset(reason);
             state.record_event(H2Event::ResetSent {
                 connection_id,
                 stream_id,
                 reason,
             });
-885
+        }
         H2StreamScript::Respond(response) => {
         H2StreamScript::Respond(response)
         | 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| {
                     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(());
-909
+            }
-910
             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 {
                         connection_id,

↧ Expand context

                         stream_id,
                     });
-920
+                }
                 H2BodyPlan::Gated {
                     before,
                     gate,
                     after,
                 } => {
                     if !before.is_empty() {
                         send.send_data(before, false).map_err(|err| {
                             H2HarnessError::new(format!(
                                 "failed to send H2 response body prefix: {err}"
                             ))
                         })?;
-932
+                    }
                     gate.wait().await.map_err(|err| {
                         H2HarnessError::new(format!("H2 response gate failed: {err}"))
                     })?;
                     send.send_data(after, true).map_err(|err| {
                         H2HarnessError::new(format!(

@@ -930,966 +989,1051 @@

↥ Expand context

                     });
-967
+                }
                 H2BodyPlan::AwaitClientReset {
                     before,
                     expected_reason,
                 } => {
                     if !before.is_empty() {
                         send.send_data(before, false).map_err(|err| {
                             H2HarnessError::new(format!(
                                 "failed to send H2 response body prefix: {err}"
                             ))
                         })?;
-978
+                    }
                     let reason = poll_fn(|cx| send.poll_reset(cx)).await.map_err(|err| {
                         H2HarnessError::new(format!("failed while awaiting client reset: {err}"))
                     })?;
                     if reason != expected_reason {
                         return Err(H2HarnessError::new(format!(
                             "expected client reset {expected_reason:?}, got {reason:?}"
                         )));

-986
+                    }
                     state.record_event(H2Event::ClientResetObserved {
                         connection_id,
                         stream_id,
                         reason,
                     });
-992
+                }
-993
+            }
-994
+        }
-995
+    }
     receive_request_body(held_request).await?;
     Ok(())
-+
+}
-+
 /// Waits for the test, then consumes a retained request body through end-of-stream.
 async fn receive_request_body(
     held_request: Option<(RecvStream, GateWaiter)>,
 ) -> Result<(), H2HarnessError> {
     let Some((mut request_body, receive)) = held_request else {
         return Ok(());
     };
     receive
         .wait()
         .await
         .map_err(|err| H2HarnessError::new(format!("H2 request-body gate failed: {err}")))?;
     while let Some(chunk) = request_body.data().await {
         let chunk = chunk.map_err(|err| {
             H2HarnessError::new(format!("failed while receiving H2 request body: {err}"))
         })?;
         request_body
             .flow_control()
             .release_capacity(chunk.len())
             .map_err(|err| {
                 H2HarnessError::new(format!("failed to release H2 request capacity: {err}"))
             })?;
-+
+    }
     Ok(())
-1023
+}
-1024
 struct ObservedIo<T> {
     inner: T,
     observer: FrameObserver,
-1028
+}
-1029
 impl<T> ObservedIo<T> {
     fn new(inner: T, connection_id: H2ConnectionId, state: Arc<SharedState>) -> Self {

↧ Expand context

         Self {
             inner,
             observer: FrameObserver::new(connection_id, state),
-1035
+        }
-1036
+    }
-1037
+}
-1038
 impl<T: AsyncRead + Unpin> AsyncRead for ObservedIo<T> {
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
         Pin::new(&mut self.inner).poll_read(cx, buf)
-1046
+    }
-1047
+}
-1048
 impl<T: AsyncWrite + Unpin> AsyncWrite for ObservedIo<T> {
     fn poll_write(
         mut self: Pin<&mut Self>,

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

@@ -1,1 +91,118 @@

↥ Expand context

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

-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;
-14
 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_http_client::Builder;
 use aws_smithy_runtime_api::client::connection::{
     CaptureSmithyConnection, ConnectionMetadata as SmithyConnectionMetadata,
 };
 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, HyperUtilLegacyPool};
 use common::h2::{
     H2BodyPlan, H2ConnectionId, H2ConnectionPlan, H2ConnectionScript, H2Event, H2Response,
     H2StreamScript, H2TestServer,
 };
 use common::tls as test_tls;
 use h2::Reason;
 use http_body_util::BodyExt;
 use std::error::Error;
-38
 trait HttpsClientBackend {
     fn build_https(
         &self,
         config: BackendConfig,
         provider: tls::Provider,
         tls_context: tls::TlsContext,
     ) -> SharedHttpClient;
-46
+}
-47
 impl HttpsClientBackend for HyperUtilLegacyPool {
     fn build_https(
         &self,
         config: BackendConfig,
         provider: tls::Provider,
         tls_context: tls::TlsContext,
     ) -> SharedHttpClient {
         let mut builder = Builder::new();
         if let Some(pool_idle_timeout) = config.pool_idle_timeout {
             builder = builder.pool_idle_timeout(pool_idle_timeout);
-58
+        }
         builder
             .tls_provider(provider)
             .tls_context(tls_context)
             .build_https()
-63
+    }
-64
+}
-65
 #[derive(Clone, Copy, Debug)]
 struct PartitionedConnectionPool;
-+
 impl HttpsClientBackend for PartitionedConnectionPool {
     fn build_https(
         &self,
         config: BackendConfig,
         provider: tls::Provider,
         tls_context: tls::TlsContext,
     ) -> SharedHttpClient {
         let mut builder = ConnectionPool::builder().tls_provider(provider);
         if let Some(pool_idle_timeout) = config.pool_idle_timeout {
             builder = builder.idle_timeout(pool_idle_timeout);
-+
+        }
         let pool = builder
             .tls_context(tls_context)
             .build_https()
             .expect("valid connection-pool config");
         let client = PoolClient::new(&pool).expect("anonymous partition exists");
         SharedHttpClient::new(client)
-+
+    }
-+
+}
-+
 fn rustls_aws_lc() -> tls::Provider {
     tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc)
-91
+}
-92
 fn h2_client_with_provider(
     backend: &dyn HttpsClientBackend,
     provider: tls::Provider,
 ) -> SharedHttpClient {
     backend.build_https(
         BackendConfig::default(),

↧ Expand context

         provider,
         test_tls::server_tls_context(),
-101
+    )
-102
+}
-103
 fn h2_client(backend: &dyn HttpsClientBackend) -> SharedHttpClient {
     h2_client_with_provider(backend, rustls_aws_lc())
-106
+}
-107
 fn stream_connection_ids(server: &H2TestServer, path: &str) -> Vec<H2ConnectionId> {
     server
         .events()
         .into_iter()
         .filter_map(|event| match event {
             H2Event::StreamAccepted {
                 connection_id,
                 path: event_path,
                 ..
             } if event_path == path => Some(connection_id),
             _ => None,

@@ -130,157 +241,278 @@

↥ Expand context

     /// Sequential streams for one origin reuse an established H2 connection.
     async fn sequential_requests_reuse_connection(backend: &dyn HttpsClientBackend) {
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([H2ConnectionScript::new()
                 .fallback(H2StreamScript::respond(H2Response::ok("hello")))]))
             .start()
             .await
             .expect("H2 server should start");
         let client = h2_client(backend);
         let connector = test_client::connector(&client);
-167
         for request_number in 1..=5 {
             let (status, body) = test_client::get_and_collect(&connector, &server.url("/")).await;
             assert_eq!(status, 200, "request {request_number}");
             assert_eq!(body, b"hello", "request {request_number}");
-172
+        }
-173
         assert_eq!(server.connection_count(), 1);
         assert_eq!(server.stream_count(), 5);
-176

         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-180
+    }
-181
     #[tokio::test]
     async fn test_sequential_requests_reuse_connection_with_hyper_util_legacy_pool() {
         sequential_requests_reuse_connection(&HyperUtilLegacyPool).await;
-185
+    }
-186
     #[tokio::test]
     async fn test_sequential_requests_reuse_connection_with_partitioned_pool() {
         sequential_requests_reuse_connection(&PartitionedConnectionPool).await;
-+
+    }
-+
     /// Concurrent requests multiplex as independent streams on a warmed H2 connection.
     async fn concurrent_requests_multiplex_on_warmed_connection(backend: &dyn HttpsClientBackend) {
         let body_gate = ManualGate::new();
         let script = H2ConnectionScript::new()
             .route("/warm", H2StreamScript::respond(H2Response::ok("warm")))
             .route(
                 "/concurrent",
                 H2StreamScript::respond(H2Response::new(http_1x::StatusCode::OK).body(
                     H2BodyPlan::gated(Bytes::new(), body_gate.waiter(), "multiplexed"),
                 )),
             );
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([script]))
             .start()
             .await
             .expect("H2 server should start");
         let client = h2_client(backend);
         let connector = test_client::connector(&client);
-210
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/warm")).await;
         assert_eq!((status, body.as_slice()), (200, b"warm".as_slice()));
-213
         let mut requests = tokio::task::JoinSet::new();
         for _ in 0..4 {
             let connector = connector.clone();
             let url = server.url("/concurrent");
             requests.spawn(async move { test_client::get_and_collect(&connector, &url).await });
-219
+        }
-220
         body_gate
             .wait_for_arrivals(4, test_client::WAIT)
             .await
             .expect("all concurrent H2 streams should reach their body gate");
         assert_eq!(server.connection_count(), 1);
         assert_eq!(server.stream_count(), 5);
         body_gate.release();
-228
         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"multiplexed".as_slice()));
-232
+        }
-233
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-237
+    }
-238
     #[tokio::test]
     async fn test_concurrent_requests_multiplex_on_warmed_connection_with_hyper_util_legacy_pool() {
         concurrent_requests_multiplex_on_warmed_connection(&HyperUtilLegacyPool).await;
-242
+    }
-243
     #[tokio::test]
     async fn test_concurrent_requests_multiplex_on_warmed_connection_with_partitioned_pool() {
         concurrent_requests_multiplex_on_warmed_connection(&PartitionedConnectionPool).await;
-+
+    }
-+
     /// Concurrent cold-start requests converge on one established H2 connection even when
     /// connection attempts race.
     async fn concurrent_cold_start_converges_on_one_h2_connection(
         backend: &dyn HttpsClientBackend,
     ) {
         let body_gate = ManualGate::new();
         let script = H2ConnectionScript::new()
             .allow_handshake_abandonment()
             .fallback(H2StreamScript::respond(
                 H2Response::new(http_1x::StatusCode::OK).body(H2BodyPlan::gated(

↧ Expand context

                     Bytes::new(),
                     body_gate.waiter(),
                     "cold",
                 )),
             ));
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::unbounded(script))
             .start()
             .await
             .expect("H2 server should start");
         let client = h2_client(backend);
         let connector = test_client::connector(&client);
         let mut requests = tokio::task::JoinSet::new();
-272
         for _ in 0..4 {
             let connector = connector.clone();
             let url = server.url("/cold");
             requests.spawn(async move { test_client::get_and_collect(&connector, &url).await });
-277
+        }
-278

@@ -251,288 +350,397 @@

↥ Expand context

                 _ => None,
             })
             .collect::<Vec<_>>();
         assert_eq!(ready_connections.len(), 1);
         let stream_connections = stream_connection_ids(&server, "/cold");
         assert_eq!(stream_connections.len(), 4);
         assert!(
             stream_connections
                 .iter()
                 .all(|connection_id| *connection_id == ready_connections[0]),
             "all cold-start requests should converge on one established H2 connection"
         );
         body_gate.release();
-301
         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()));
-305
+        }
         assert_eq!(server.stream_count(), 4);
-307

         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-311
+    }
-312
     #[tokio::test]
     async fn test_concurrent_cold_start_converges_on_one_h2_connection_with_hyper_util_legacy_pool()
-315
+    {
         concurrent_cold_start_converges_on_one_h2_connection(&HyperUtilLegacyPool).await;
-317
+    }
-+
     #[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;
-+
+    }
-323
+}
-324
 mod connection_metadata {
     use super::*;
-327
     /// 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");
         let client = h2_client(backend);
         let connector = test_client::connector(&client);
-339
         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();
-344
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/second")).await;
         assert_eq!((status, body.as_slice()), (200, b"ok".as_slice()));
-347
         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);
-352
         drop(metadata);
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-357
+    }
-358
     #[tokio::test]
     async fn test_poisoned_connection_is_not_reused_with_hyper_util_legacy_pool() {
         poisoned_connection_is_not_reused(&HyperUtilLegacyPool).await;
-362
+    }
-+
     #[tokio::test]
     async fn test_poisoned_connection_is_not_reused_with_partitioned_pool() {
         poisoned_connection_is_not_reused(&PartitionedConnectionPool).await;
-+
+    }
-368
+}
-369
 mod stream_failures {
     use super::*;
-372
     /// A peer reset fails only its H2 stream and leaves the connection available for reuse.
     async fn stream_reset_does_not_retire_connection(backend: &dyn HttpsClientBackend) {
         let reset_gate = ManualGate::new();
         let script = H2ConnectionScript::new()
             .route(

↧ Expand context

                 "/reset",
                 H2StreamScript::respond(H2Response::new(http_1x::StatusCode::OK).body(
                     H2BodyPlan::reset_after("partial", reset_gate.waiter(), Reason::CANCEL),
                 )),
-382
+            )
             .route("/ok", H2StreamScript::respond(H2Response::ok("reused")));
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([script]))
             .start()
             .await
             .expect("H2 server should start");
         let client = h2_client(backend);
         let connector = test_client::connector(&client);
-391
         let response = test_client::send_request(
             &connector,
             HttpRequest::get(server.url("/reset")).expect("valid HTTP request"),
-395
+        )
         .await
         .expect("response headers should succeed before the stream reset");

@@ -371,418 +430,482 @@

↥ Expand context

                 matches!(
                     event,
                     H2Event::ResetSent {
                         reason: Reason::CANCEL,
                         ..
-423
+                    }
-424
+                )
             })
             .await
             .expect("server should send the scripted stream reset");
         assert!(matches!(reset_event, H2Event::ResetSent { .. }));
-429
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/ok")).await;
         assert_eq!((status, body.as_slice()), (200, b"reused".as_slice()));
         assert_eq!(
             single_stream_connection(&server, "/reset"),
             single_stream_connection(&server, "/ok")
         );
         assert_eq!(server.connection_count(), 1);
-437

         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-441
+    }
-442
     #[tokio::test]
     async fn test_stream_reset_does_not_retire_connection_with_hyper_util_legacy_pool() {
         stream_reset_does_not_retire_connection(&HyperUtilLegacyPool).await;
-446
+    }
-447
     #[tokio::test]
     async fn test_stream_reset_does_not_retire_connection_with_partitioned_pool() {
         stream_reset_does_not_retire_connection(&PartitionedConnectionPool).await;
-+
+    }
-+
     /// Dropping an incomplete response body cancels only that stream and permits reuse.
     async fn dropping_response_body_cancels_only_stream(backend: &dyn HttpsClientBackend) {
         let script = H2ConnectionScript::new()
             .route(
                 "/drop",
                 H2StreamScript::respond(
                     H2Response::new(http_1x::StatusCode::OK)
                         .body(H2BodyPlan::await_client_reset("partial", Reason::CANCEL)),
                 ),
-462
+            )

↧ Expand context

             .route("/ok", H2StreamScript::respond(H2Response::ok("reused")));
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([script]))
             .start()
             .await
             .expect("H2 server should start");
         let client = h2_client(backend);
         let connector = test_client::connector(&client);
-471
         let response = test_client::send_request(
             &connector,
             HttpRequest::get(server.url("/drop")).expect("valid HTTP request"),
-475
+        )
         .await
         .expect("response headers should succeed");
         assert_eq!(response.status().as_u16(), 200);
         let mut body = response.into_body();
         let frame = tokio::time::timeout(test_client::WAIT, body.frame())
             .await
             .expect("response body frame should arrive within the outer deadline")

@@ -439,491 +498,555 @@

↥ Expand context

         server
             .wait_for_event(test_client::WAIT, |event| {
                 matches!(
                     event,
                     H2Event::ClientResetObserved {
                         reason: Reason::CANCEL,
                         ..
-498
+                    }
-499
+                )
             })
             .await
             .expect("dropping the body should reset only its H2 stream");
-503
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/ok")).await;
         assert_eq!((status, body.as_slice()), (200, b"reused".as_slice()));
         assert_eq!(
             single_stream_connection(&server, "/drop"),
             single_stream_connection(&server, "/ok")
         );
         assert_eq!(server.connection_count(), 1);

-511
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-515
+    }
-516
     #[tokio::test]
     async fn test_dropping_response_body_cancels_only_stream_with_hyper_util_legacy_pool() {
         dropping_response_body_cancels_only_stream(&HyperUtilLegacyPool).await;
-520
+    }
-+
     #[tokio::test]
     async fn test_dropping_response_body_cancels_only_stream_with_partitioned_pool() {
         dropping_response_body_cancels_only_stream(&PartitionedConnectionPool).await;
-+
+    }
-526
+}
-527
 mod goaway_and_replacement {
     use super::*;
-530
     /// Graceful GOAWAY drains an eligible in-flight stream while later streams use a
     /// replacement connection.
     async fn graceful_goaway_preserves_in_flight_stream_and_replaces_connection(
         backend: &dyn HttpsClientBackend,
     ) {

↧ Expand context

         let held_body_gate = ManualGate::new();
         let script = H2ConnectionScript::new()
             .route(
                 "/held",
                 H2StreamScript::respond(H2Response::new(http_1x::StatusCode::OK).body(
                     H2BodyPlan::gated("held-", held_body_gate.waiter(), "complete"),
                 )),
-543
+            )
             .route(
                 "/after",
                 H2StreamScript::respond(H2Response::ok("replacement")),
             );
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([script.clone(), script]))
             .start()
             .await
             .expect("H2 server should start");
         let client = h2_client(backend);
         let connector = test_client::connector(&client);
-555

@@ -556,613 +720,884 @@

↥ Expand context

         }));
-614
         held_body_gate.release();
         let (status, body) = test_client::collect_response(held_response).await;
         assert_eq!(
             (status, body.as_slice()),
             (200, b"held-complete".as_slice())
         );
         server
             .wait_for_event(test_client::WAIT, |event| {
                 matches!(
                     event,
                     H2Event::ConnectionClosed { connection_id, reason: ConnectionCloseReason::ClientClosed }
                         if *connection_id == original_connection
-627
+                )
             })
             .await
             .expect("the original connection should close after its held stream completes");
-631
         drop(connector);

         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-635
+    }
-636
     #[tokio::test]
     async fn test_graceful_goaway_preserves_in_flight_stream_and_replaces_connection_with_hyper_util_legacy_pool(
     ) {
         graceful_goaway_preserves_in_flight_stream_and_replaces_connection(&HyperUtilLegacyPool)
             .await;
-642
+    }
-+
     #[tokio::test]
     async fn test_graceful_goaway_preserves_in_flight_stream_and_replaces_connection_with_partitioned_pool(
     ) {
         graceful_goaway_preserves_in_flight_stream_and_replaces_connection(
             &PartitionedConnectionPool,
-+
+        )
         .await;
-+
+    }
-+
     #[tokio::test]
     async fn goaway_before_first_stream_has_bounded_reacquisition() {
         let script = H2ConnectionScript::new()
             .goaway_on_ready()
             .fallback(H2StreamScript::respond(H2Response::ok("unexpected")));
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::unbounded(script))
             .start()
             .await
             .expect("H2 server should start");
         let client = h2_client(&PartitionedConnectionPool);
         let connector = test_client::connector(&client);
-+
         let outcome = tokio::time::timeout(
             test_client::WAIT,
             test_client::send_request(
                 &connector,
                 HttpRequest::get(server.url("/before-first-stream")).expect("valid HTTP request"),
             ),
-+
+        )
         .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()
         );
-+
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-+
+    }
-689
+}
-690
 #[cfg(feature = "s2n-tls")]
 mod protocol_negotiation {
     use super::*;
-694
     /// 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()
             .await
             .expect("H2 server should start");
         let client = h2_client_with_provider(backend, tls::Provider::S2nTls);
         let connector = test_client::connector(&client);
-705
         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}");
-710
+        }
-711
         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())]);
         assert_eq!(server.connection_count(), 1);
         assert_eq!(server.stream_count(), 3);
-723
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-727
+    }
-728
     #[tokio::test]
     async fn test_s2n_negotiates_h2_and_reuses_connection_with_hyper_util_legacy_pool() {
         s2n_negotiates_h2_and_reuses_connection(&HyperUtilLegacyPool).await;
-732
+    }
-+
     #[tokio::test]
     async fn test_s2n_negotiates_h2_and_reuses_connection_with_partitioned_pool() {
         s2n_negotiates_h2_and_reuses_connection(&PartitionedConnectionPool).await;
-+
+    }
-738
+}
-739
 mod idle_timeout {
     use super::*;
     use aws_smithy_types::body::SdkBody;
     use http_body_1x::{Body, Frame, SizeHint};
     use std::convert::Infallible;
     use std::future::Future;
     use std::pin::Pin;
     use std::task::{Context, Poll};
     use std::time::Duration;
     use tokio::sync::oneshot;
-750
     const IDLE_TIMEOUT: Duration = Duration::from_millis(100);
-752
     /// Streaming request body that remains open until the test releases it.
     struct HeldUpload {
         finish: oneshot::Receiver<()>,
         complete: bool,
-+
+    }
-+
     impl Body for HeldUpload {
         type Data = Bytes;
         type Error = Infallible;
-+
         fn poll_frame(
             mut self: Pin<&mut Self>,
             cx: &mut Context<'_>,
         ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
             if self.complete {
                 return Poll::Ready(None);
-+
+            }
             match Pin::new(&mut self.finish).poll(cx) {
                 Poll::Ready(_) => {
                     self.complete = true;
                     Poll::Ready(None)
-+
+                }
                 Poll::Pending => Poll::Pending,
-+
+            }
-+
+        }
-+
         fn is_end_stream(&self) -> bool {
             self.complete
-+
+        }
-+
         fn size_hint(&self) -> SizeHint {
             SizeHint::default()
-+
+        }
-+
+    }
-+
     fn held_upload() -> (oneshot::Sender<()>, SdkBody) {
         let (finish, finished) = oneshot::channel();
         let body = HeldUpload {
             finish: finished,
             complete: false,
         };
         (finish, SdkBody::from_body_1_x(body))
-+
+    }
-+
     fn client_with_idle_timeout(backend: &dyn HttpsClientBackend) -> SharedHttpClient {
         backend.build_https(
             BackendConfig {
                 pool_idle_timeout: Some(IDLE_TIMEOUT),
             },
             rustls_aws_lc(),
             test_tls::server_tls_context(),
-804
+        )
-805
+    }
-806
     /// An idle H2 connection is closed after its configured timeout and then replaced.
     async fn idle_connection_is_evicted_after_timeout(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");
         let client = client_with_idle_timeout(backend);
         let connector = test_client::connector(&client);
-818
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/first")).await;
         assert_eq!((status, body.as_slice()), (200, b"ok".as_slice()));
         let first_connection = single_stream_connection(&server, "/first");
         server
             .wait_for_event(test_client::WAIT, |event| {
                 matches!(
                     event,
                     H2Event::ConnectionClosed { connection_id, reason: ConnectionCloseReason::ClientClosed }
                         if *connection_id == first_connection
-828
+                )
             })
             .await
             .expect("the idle H2 connection should close after its pool timeout");
-832
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/second")).await;
         assert_eq!((status, body.as_slice()), (200, b"ok".as_slice()));
         assert_ne!(
             first_connection,
             single_stream_connection(&server, "/second")
         );
-839
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-843
+    }
-844
     #[tokio::test]
     async fn test_idle_connection_is_evicted_after_timeout_with_hyper_util_legacy_pool() {
         idle_connection_is_evicted_after_timeout(&HyperUtilLegacyPool).await;
-848
+    }
-849
     #[tokio::test]
     async fn test_idle_connection_is_evicted_after_timeout_with_partitioned_pool() {
         idle_connection_is_evicted_after_timeout(&PartitionedConnectionPool).await;
-+
+    }
-+
     /// An active stream survives the idle deadline, but the connection is replaced after the
     /// stream completes.
     async fn active_stream_survives_idle_timeout_but_later_request_uses_replacement(
         backend: &dyn HttpsClientBackend,
     ) {
         let held_body_gate = ManualGate::new();
         let script = H2ConnectionScript::new()
             .route(
                 "/held",
                 H2StreamScript::respond(H2Response::new(http_1x::StatusCode::OK).body(

↧ Expand context

                     H2BodyPlan::gated("held-", held_body_gate.waiter(), "complete"),
                 )),
-867
+            )
             .route("/second", H2StreamScript::respond(H2Response::ok("second")));
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([script.clone(), script]))
             .start()
             .await
             .expect("H2 server should start");
         let client = client_with_idle_timeout(backend);
         let connector = test_client::connector(&client);
-876
         let held_response = test_client::send_request(
             &connector,
             HttpRequest::get(server.url("/held")).expect("valid HTTP request"),
-880
+        )
         .await
         .expect("held response headers should succeed");
-883
         held_body_gate

@@ -731,895 +761,1129 @@

↥ Expand context

                 H2Event::ConnectionClosed { connection_id, .. }
                     if *connection_id == first_connection
-897
+            )
         }));
-899
         held_body_gate.release();
         let (status, body) = test_client::collect_response(held_response).await;
         assert_eq!(
             (status, body.as_slice()),
             (200, b"held-complete".as_slice())
         );
-906
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/second")).await;
         assert_eq!((status, body.as_slice()), (200, b"second".as_slice()));
         let second_connection = single_stream_connection(&server, "/second");
         assert_ne!(first_connection, second_connection);
-911
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");

-915
+    }
-916
     #[tokio::test]
     async fn test_active_stream_survives_idle_timeout_but_later_request_uses_replacement_with_hyper_util_legacy_pool(
     ) {
         active_stream_survives_idle_timeout_but_later_request_uses_replacement(
             &HyperUtilLegacyPool,
-922
+        )
         .await;
-924
+    }
-+
     #[tokio::test]
     async fn test_active_stream_survives_idle_timeout_but_later_request_uses_replacement_with_partitioned_pool(
     ) {
         active_stream_survives_idle_timeout_but_later_request_uses_replacement(
             &PartitionedConnectionPool,
-+
+        )
         .await;
-+
+    }
-+
     /// Idle expiration retains a physical connection while its upload remains active.
     #[tokio::test]
     async fn response_completion_does_not_close_active_upload_at_idle_timeout() {
         let held_request_gate = ManualGate::new();
         let script = H2ConnectionScript::new()
             .route(
                 "/upload",
                 H2StreamScript::respond_before_receiving_request_body(
                     H2Response::ok("early response"),
                     held_request_gate.waiter(),
                 ),
-+
+            )
             .route("/reuse", H2StreamScript::respond(H2Response::ok("reused")));
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([script.clone(), script]))
             .start()
             .await
             .expect("H2 server should start");
         let client = client_with_idle_timeout(&PartitionedConnectionPool);
         let connector = test_client::connector(&client);
-+
         let (finish_upload, upload) = held_upload();
         let mut request = HttpRequest::new(upload);
         request.set_method("POST").expect("valid HTTP method");
         request
             .set_uri(server.url("/upload"))
             .expect("valid HTTP URI");
         let response = test_client::send_request(&connector, request)
             .await
             .expect("response should arrive before upload completion");
         let (status, body) = test_client::collect_response(response).await;
         assert_eq!(
             (status, body.as_slice()),
             (200, b"early response".as_slice())
         );
-+
         held_request_gate
             .wait_until_reached(test_client::WAIT)
             .await
             .expect("server should retain the active request body");
         let upload_connection = single_stream_connection(&server, "/upload");
         tokio::time::sleep(IDLE_TIMEOUT * 2).await;
         assert!(
             !server.events().iter().any(|event| {
                 matches!(
                     event,
                     H2Event::ConnectionClosed { connection_id, .. }
                         if *connection_id == upload_connection
-+
+                )
             }),
             "idle expiration closed a connection with an active upload"
         );
-+
         let (status, body) = test_client::get_and_collect(&connector, &server.url("/reuse")).await;
         assert_eq!((status, body.as_slice()), (200, b"reused".as_slice()));
         assert_ne!(
             upload_connection,
             single_stream_connection(&server, "/reuse"),
             "an expired HTTP/2 generation accepted a new request"
         );
         assert_eq!(server.connection_count(), 2);
-+
         finish_upload
             .send(())
             .expect("request body disappeared before upload completion");
         held_request_gate.release();
         server
             .wait_for_event(test_client::WAIT, |event| {
                 matches!(
                     event,
                     H2Event::ConnectionClosed { connection_id, .. }
                         if *connection_id == upload_connection
-+
+                )
             })
             .await
             .expect("draining connection should close after upload completion");
         drop(connector);
         drop(client);
         server.shutdown().await.expect("clean H2 server shutdown");
-+
+    }
-+
+}
-+
 mod partition_reuse {
     use super::*;
     use std::time::Duration;
-+
     fn partitioned_clients(
         scope: ConnectionReuseScope,
     ) -> (ConnectionPool, SharedHttpConnector, SharedHttpConnector) {
         let first = PartitionId::from_index(1);
         let second = PartitionId::from_index(2);
         let pool = ConnectionPool::builder()
             .tls_provider(rustls_aws_lc())
             .tls_context(test_tls::server_tls_context())
             .partitions([
                 Partition::new(first, TokioDriverSpawner::current()),
                 Partition::new(second, TokioDriverSpawner::current()),
             ])
             .connection_reuse_scope(scope)
             .max_connections_per_host(1)
             .build_https()
             .expect("valid partitioned HTTPS pool");
         let first_client = SharedHttpClient::new(
             PoolClient::from_partition(&pool, first).expect("first partition should resolve"),
         );
         let second_client = SharedHttpClient::new(
             PoolClient::from_partition(&pool, second).expect("second partition should resolve"),
         );
-+
+        (
             pool,
             test_client::connector(&first_client),
             test_client::connector(&second_client),
-+
+        )
-+
+    }
-+
     async fn eligible_partition_reuses_peer_h2(scope: ConnectionReuseScope) {
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([H2ConnectionScript::new()
                 .fallback(H2StreamScript::respond(H2Response::ok("shared")))]))
             .start()
             .await
             .expect("H2 server should start");
         let (pool, first, second) = partitioned_clients(scope);
-+
         let first_result = test_client::get_and_collect(&first, &server.url("/first")).await;
         let second_result = test_client::get_and_collect(&second, &server.url("/second")).await;
-+
         assert_eq!(first_result, (200, b"shared".to_vec()));
         assert_eq!(second_result, (200, b"shared".to_vec()));
         assert_eq!(server.connection_count(), 1);
         assert_eq!(
             single_stream_connection(&server, "/first"),
             single_stream_connection(&server, "/second"),
             "eligible partitions should dispatch through the connection-owning generation"
         );
-+
         drop(first);
         drop(second);
         drop(pool);
         server.shutdown().await.expect("clean H2 server shutdown");
-+
+    }
-+
     #[tokio::test]
     async fn pool_scope_reuses_a_peer_h2_generation() {
         eligible_partition_reuses_peer_h2(ConnectionReuseScope::Pool).await;
-+
+    }
-+
     #[tokio::test]
     async fn matching_network_interface_scope_reuses_a_peer_h2_generation() {
         eligible_partition_reuses_peer_h2(ConnectionReuseScope::NetworkInterface).await;
-+
+    }
-+
     #[tokio::test]
     async fn partition_scope_waits_until_out_of_scope_h2_releases_capacity() {
         let script = H2ConnectionScript::new()
             .fallback(H2StreamScript::respond(H2Response::ok("partition")));
         let server = H2TestServer::builder()
             .connections(H2ConnectionPlan::queue([script.clone(), script]))
             .start()
             .await
             .expect("H2 server should start");
         let (pool, first, second) = partitioned_clients(ConnectionReuseScope::Partition);
-+
         let (status, body, metadata) =
             get_and_collect_with_capture(&first, &server.url("/first")).await;
         assert_eq!((status, body.as_slice()), (200, b"partition".as_slice()));
         let first_connection = single_stream_connection(&server, "/first");
-+
         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);
-+
         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);
-+
         drop(metadata);
         drop(first);
         drop(pool);
         server.shutdown().await.expect("clean H2 server shutdown");
-+
+    }
-1129
+}

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

@@ -1,1 +88,94 @@

↥ Expand context

  * SPDX-License-Identifier: Apache-2.0
  */
-5
 #![cfg(any(feature = "__rustls", feature = "s2n-tls",))]
-7
 mod common {
     pub(crate) mod tls;
-10
+}
-11
 use aws_smithy_async::time::SystemTimeSource;
 #[cfg(all(feature = "rustls-aws-lc", feature = "rt-tokio"))]
 use aws_smithy_http_client::pool::{Client as PoolClient, ConnectionPool};
 use aws_smithy_http_client::tls;
 #[cfg(any(feature = "rustls-aws-lc", feature = "s2n-tls"))]
 use aws_smithy_http_client::tls::{ServerName, TlsContext};
 use aws_smithy_runtime_api::box_error::BoxError;
 use aws_smithy_runtime_api::client::http::{HttpClient, HttpConnector, HttpConnectorSettings};
 use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
 use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
 use aws_smithy_types::byte_stream::ByteStream;
 use common::tls as test_tls;
 use http_1x::{Method, Request, Response, StatusCode};
 use http_body_util::{BodyExt, Full};
 use hyper::body::{Bytes, Incoming};
 use hyper::service::service_fn;
 use hyper_util::rt::{TokioExecutor, TokioIo};
 use hyper_util::server::conn::auto::Builder;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::net::TcpListener;
 use tokio::task::JoinHandle;
 use tracing::{debug, error};
-36
 struct TestServer {
     _handle: JoinHandle<()>,
     listen_addr: SocketAddr,
     conn_count: Arc<()>,
-41
+}
-42
 impl TestServer {
     /// Return the number of active connections to this server
     fn conn_count(&self) -> usize {
         // 1 reference for the struct MockProxyServer, 1 reference for the
         // socket task.
         Arc::strong_count(&self.conn_count)
             .checked_sub(2)
             .expect("de-count 2 refs")
-51
+    }
-52
+}
-53
 async fn server() -> Result<TestServer, BoxError> {
     server_with_alpn(&[b"h2", b"http/1.1", b"http/1.0"]).await
-+
+}
-+
 async fn server_with_alpn(alpn_protocols: &[&[u8]]) -> Result<TestServer, BoxError> {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let addr = listener.local_addr().unwrap();
-61
     debug!("Starting to serve on https://{}", addr);
-63
     let tls_acceptor = test_tls::server_tls_acceptor(&[b"h2", b"http/1.1", b"http/1.0"])?;
     let tls_acceptor = test_tls::server_tls_acceptor(alpn_protocols)?;
     let service = service_fn(echo);
-66
     let conn_count = Arc::new(());
     let server_conn_count = conn_count.clone();
-69
     let server = async move {
         loop {
             let (tcp_stream, remote_addr) = listener.accept().await.unwrap();
             debug!("accepted connection from: {}", remote_addr);
-74

↧ Expand context

             let tls_acceptor = tls_acceptor.clone();
             let connection_conn_count = server_conn_count.clone();
             tokio::spawn(async move {
                 let _connection_conn_count = connection_conn_count;
                 let tls_stream = match tls_acceptor.accept(tcp_stream).await {
                     Ok(tls_stream) => tls_stream,
                     Err(err) => {
                         error!("failed to perform tls handshake: {err:#}");
                         return;
-84
+                    }
                 };
                 if let Err(err) = Builder::new(TokioExecutor::new())
                     .serve_connection(TokioIo::new(tls_stream), service)
                     .await
-89
+                {
                     error!("failed to serve connection: {err:#}");
-91
+                }
             });
-93
+        }
     };

@@ -116,122 +175,252 @@

↥ Expand context

     };
     Ok(response)
-124
+}
-125
 #[cfg(feature = "rustls-aws-lc")]
 #[should_panic(expected = "InvalidCertificate(UnknownIssuer)")]
 #[tokio::test]
 async fn test_rustls_aws_lc_native_ca() {
     let client = aws_smithy_http_client::Builder::new()
         .tls_provider(tls::Provider::Rustls(
             tls::rustls_provider::CryptoMode::AwsLc,
         ))
         .build_https();
-135
     run_tls_test(&client).await.unwrap()
-137
+}
-138
 #[cfg(feature = "rustls-aws-lc")]
 #[tokio::test]
 async fn test_rustls_aws_lc_custom_ca() {

     let client = aws_smithy_http_client::Builder::new()
         .tls_provider(tls::Provider::Rustls(
             tls::rustls_provider::CryptoMode::AwsLc,
         ))
         .tls_context(test_tls::server_tls_context())
         .build_https();
-148
     run_tls_test(&client).await.unwrap()
-150
+}
-151
 #[cfg(all(feature = "rustls-aws-lc", feature = "rt-tokio"))]
 #[tokio::test]
 async fn partitioned_pool_falls_back_to_h1_after_alpn() {
     let server = server_with_alpn(&[b"http/1.1"]).await.unwrap();
     let pool = ConnectionPool::builder()
         .tls_provider(tls::Provider::Rustls(
             tls::rustls_provider::CryptoMode::AwsLc,
         ))
         .tls_context(test_tls::server_tls_context())
         .build_https()
         .expect("valid HTTPS pool");
     let client = PoolClient::new(&pool).expect("anonymous partition exists");
     let connector_settings = HttpConnectorSettings::builder().build();
     let runtime_components = RuntimeComponentsBuilder::for_tests()
         .with_time_source(Some(SystemTimeSource::new()))
         .build()
         .unwrap();
     let connector = client.http_connector(&connector_settings, &runtime_components);
     let endpoint = format!("https://localhost:{}/", server.listen_addr.port());
-+
     for _ in 0..2 {
         let mut response = connector
             .call(HttpRequest::get(&endpoint).unwrap())
             .await
             .expect("HTTP/1 request over TLS should succeed");
         let body = ByteStream::new(response.take_body())
             .collect()
             .await
             .expect("response body should be readable")
             .into_bytes();
         assert_eq!(b"Hello TLS!", &body[..]);
-+
+    }
     assert_eq!(1, server.conn_count());
-+
+}
-+
 #[cfg(all(feature = "rustls-aws-lc", feature = "rt-tokio"))]
 #[tokio::test]
 async fn partitioned_pool_narrows_alpn_for_h1_required_request() {
     let server = server_with_alpn(&[b"h2", b"http/1.1"]).await.unwrap();
     let pool = ConnectionPool::builder()
         .tls_provider(tls::Provider::Rustls(
             tls::rustls_provider::CryptoMode::AwsLc,
         ))
         .tls_context(test_tls::server_tls_context())
         .build_https()
         .expect("valid HTTPS pool");
     let client = PoolClient::new(&pool).expect("anonymous partition exists");
     let connector_settings = HttpConnectorSettings::builder().build();
     let runtime_components = RuntimeComponentsBuilder::for_tests()
         .with_time_source(Some(SystemTimeSource::new()))
         .build()
         .unwrap();
     let connector = client.http_connector(&connector_settings, &runtime_components);
     let endpoint = format!("https://localhost:{}/", server.listen_addr.port());
     let mut request = HttpRequest::new(ByteStream::default().into_inner());
     request
         .set_method("CONNECT")
         .expect("CONNECT is a valid HTTP method");
     request
         .set_uri(endpoint.as_str())
         .expect("valid server URI");
-+
     let response = connector
         .call(request)
         .await
         .expect("HTTP/1-required request should narrow the ALPN offer");
-+
     assert_eq!(StatusCode::NOT_FOUND.as_u16(), response.status().as_u16());
     assert_eq!(1, server.conn_count());
-+
+}
-+
 #[cfg(feature = "rustls-aws-lc")]
 #[tokio::test(start_paused = false)]
 // can't have paused clock due to <https://github.com/hyperium/hyper/issues/3950>
 async fn test_rustls_aws_lc_custom_ca_with_timeout() {
     const TIMEOUT: Duration = Duration::from_secs(10);
     let client = aws_smithy_http_client::Builder::new()
         .pool_idle_timeout(TIMEOUT)
         .tls_provider(tls::Provider::Rustls(
             tls::rustls_provider::CryptoMode::AwsLc,
         ))

↧ Expand context

         .tls_context(test_tls::server_tls_context())
         .build_https();
-235
     run_tls_test_with_idle_timeout(&client, Some(TIMEOUT))
         .await
         .unwrap()
-239
+}
-240
 #[cfg(feature = "rustls-aws-lc-fips")]
 #[should_panic(expected = "InvalidCertificate(UnknownIssuer)")]
 #[tokio::test]
 async fn test_rustls_aws_lc_fips_native_ca() {
     let client = aws_smithy_http_client::Builder::new()
         .tls_provider(tls::Provider::Rustls(
             tls::rustls_provider::CryptoMode::AwsLcFips,
         ))
         .build_https();
-250
     run_tls_test(&client).await.unwrap()
-252
+}

Pages: 1 2 3 4 5 6 7

1 1	`/*`
2 2	`* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.`