AWS SDK

rev. 39e8f0a4a4cd7fc94cae1e2f796e9c21897d9cb9

Files changed:

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

@@ -176,176 +289,290 @@

-176
+            }
-177
+        }
-178
+    }
-179
+}
-180
 /// Handshakes, installs, and publishes one winning flight.
 async fn drive_flight(
     context: AcquisitionContext,
     flight: H2FlightId,
     permit: EstablishmentPermit,
     transport: ConnectedTransport,
     mut establishment: ConnectionEstablishment,
 ) {
     let mut completion = FlightCompletionGuard::new(context.cell.clone(), flight);
     let id = match next_connection_id(&context.pool) {
         Ok(id) => id,
         Err(error) => {
             let error = ConnectorError::other(Box::new(error), None);
             fail_flight_establishment(&mut completion, establishment, error);
             return;

-196
+        }
     };
     let info = ConnectionInfo::new(
         id,
         context.cell.id().origin().clone(),
         context.partition.id(),
         ConnectionProtocol::Http2,
         transport.metadata,
     );
     let (connection, physical) =
         ConnectionState::pending_open(info, context.cell.connection_stats());
         ConnectionState::pending_open(info, context.owner_spawner, context.cell.connection_stats());
     let io = ConnectionIo::new(transport.io, physical);
     let executor = PartitionExecutor {
         spawner: context.owner_spawner.clone(),
         spawner: connection.owner_spawner(),
     };
     establishment.protocol_handshake_started();
     let (sender, driver) = match hyper::client::conn::http2::Builder::new(executor)
         .handshake::<_, SdkBody>(io)
         .await
-215
+    {
         Ok(established) => established,
         Err(error) => {
             let error = downcast_error(Box::new(error));
             establishment.protocol_handshake_failed();
             connection.logical_close(CloseReason::ProtocolClosed);
             fail_flight_establishment(&mut completion, establishment, error);
             return;
-223
+        }
     };
     establishment.protocol_handshake_completed();
-226
     if let Err(lease) = connection.open(permit.into_lease()) {
         drop(lease);
         connection.logical_close(CloseReason::ProtocolClosed);
         let error = ConnectorError::io(Box::new(std::io::Error::other(
             "HTTP/2 connection closed before installation",
         )));
         fail_flight_establishment(&mut completion, establishment, error);
         return;
-235
+    }
     establishment.installed(&connection);
-237
     let installed = OriginCell::complete_h2_flight(
         &context.cell,
         flight,
         connection.clone(),
         H2Sender::from_hyper(sender),
         context.cell.idle_deadline(),
     );
     let installed = match installed {
         Ok(installed) => installed,
         Err((_connection, _sender)) => {
             connection.logical_close(CloseReason::ProtocolClosed);
             let error = ConnectorError::io(Box::new(std::io::Error::other(
                 "HTTP/2 flight became stale before installation",
             )));
             fail_flight_establishment(&mut completion, establishment, error);
             return;
-254
+        }
     };
-256
     let generation = installed;
     let driver_guard = H2DriverGuard::new(H2CloseHandle::new(&context.cell, generation));
     let driver_info = connection.info().clone();
     context.owner_spawner.spawn(Box::pin(async move {
     connection.owner_spawner().spawn(Box::pin(async move {
         if let Err(error) = driver.await {
             tracing::debug!(
                 connection_id = %driver_info.id(),
                 connection_partition = ?driver_info.owner_partition(),
                 origin_scheme = %driver_info.origin().scheme(),
                 origin_host = driver_info.origin().host(),
                 origin_port = ?driver_info.origin().port(),
                 error = ?error,
                 "HTTP/2 connection driver failed"
             );

↧ Expand context

-271
+        }
         driver_guard.protocol_closed();
     }));
-274
     completion.disarm();
     establishment.opened(&connection);
     tracing::debug!(
         connection_id = %connection.id(),
         request_partition = ?context.partition.id(),
         connection_partition = ?connection.owner_partition(),
         origin_scheme = %connection.info().origin().scheme(),
         origin_host = connection.info().origin().host(),
         origin_port = ?connection.info().origin().port(),
         h2_generation = ?generation,
         "HTTP/2 connection established"
     );
-287
+}
-288
 /// Connector classification copied to each participant in one failed flight.
 #[derive(Clone, Copy, Debug)]

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

@@ -148,148 +226,240 @@

↥ Expand context

 impl ConnectionEstablishmentInfo {
     /// Returns this establishment's pool-assigned identity.
     pub fn id(&self) -> ConnectionEstablishmentId {
         self.id
-152
+    }
-153
     /// Returns the canonical origin being connected.
     pub fn origin(&self) -> &OriginKey {
         &self.origin
-157
+    }
-158
     /// Returns the partition that owns the establishment task.
     pub fn partition(&self) -> PartitionId {
         self.partition
-162
+    }
-163
+}
-164
 /// Measurements collected during one pool-level connection establishment.
 ///
 /// A later retry receives a different [`ConnectionEstablishmentId`] and its own

 /// measurements.
 #[derive(Clone, Copy, Debug)]
 #[non_exhaustive]
 pub struct ConnectionEstablishmentStats {
     total_duration: Duration,
     transport_duration: Duration,
     protocol_handshake_duration: Option<Duration>,
-175
+}
-176
 impl ConnectionEstablishmentStats {
     /// Returns elapsed time from establishment start through its terminal event.
     /// Returns elapsed time through successful protocol installation or terminal failure.
     ///
     /// Successful measurements are frozen before the connection is published
     /// to waiting pool demand.
     pub fn total_duration(&self) -> Duration {
         self.total_duration
-184
+    }
-185
     /// Returns elapsed time spent in the configured transport connector.
     ///
     /// This may include DNS, socket, proxy, and TLS work. Custom connectors may
     /// perform a different set of transport operations.
     pub fn transport_duration(&self) -> Duration {
         self.transport_duration
-192
+    }
-193
     /// Returns elapsed time spent in the Hyper protocol handshake, when started.
     ///
     /// Failures before protocol handshake return `None`.
     pub fn protocol_handshake_duration(&self) -> Option<Duration> {
         self.protocol_handshake_duration
-199
+    }
-+
     fn connection_metadata(
         &self,
     ) -> aws_smithy_runtime_api::client::connection::ConnectionEstablishmentMetadata {
         let mut builder =
             aws_smithy_runtime_api::client::connection::ConnectionEstablishmentMetadata::builder()
                 .total_duration(self.total_duration)
                 .transport_duration(self.transport_duration);
         builder.set_protocol_handshake_duration(self.protocol_handshake_duration);
         builder.build()
-+
+    }
-211
+}
-212
 /// Stage that terminated one failed connection establishment.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[non_exhaustive]
 pub enum ConnectionEstablishmentStage {
     /// Work performed by the configured transport connector.
     Transport,
     /// Selection of an HTTP protocol compatible with the request.
     ProtocolSelection,

↧ Expand context

     /// Hyper's HTTP/1 or HTTP/2 client handshake.
     ProtocolHandshake,
     /// Transfer of a handshaken connection into pool-owned protocol state.
     PoolInstallation,
-225
+}
-226
 /// A terminal failure before a connection became available for pool dispatch.
 #[derive(Debug)]
 pub struct ConnectionEstablishmentFailed<'a> {
     establishment: &'a ConnectionEstablishmentInfo,
     stats: &'a ConnectionEstablishmentStats,
     stage: ConnectionEstablishmentStage,
     remote_addr: Option<SocketAddr>,
     protocol: Option<ConnectionProtocol>,
     error: &'a (dyn Error + Send + Sync),
-236
+}
-237
 impl<'a> ConnectionEstablishmentFailed<'a> {
     /// Returns the failed establishment's identity and routing context.
     pub fn establishment(&self) -> &'a ConnectionEstablishmentInfo {

@@ -241,255 +301,315 @@

↥ Expand context

     /// before a later stage failed.
     pub fn remote_addr(&self) -> Option<SocketAddr> {
         self.remote_addr
-258
+    }
-259
     /// Returns the selected HTTP protocol, when selection completed.
     pub fn protocol(&self) -> Option<ConnectionProtocol> {
         self.protocol
-263
+    }
-264
     /// Returns the terminal error delivered to connection acquisition.
     pub fn error(&self) -> &'a (dyn Error + Send + Sync) {
         self.error
-268
+    }
-269
+}
-270
 /// A connection installed and available for pool dispatch.
 #[derive(Debug)]
 pub struct ConnectionOpened<'a> {
     establishment: &'a ConnectionEstablishmentInfo,

     stats: &'a ConnectionEstablishmentStats,
     connection: &'a PoolArc<ConnectionInfo>,
-277
+}
-278
 impl<'a> ConnectionOpened<'a> {
     /// Returns the establishment that created this connection.
     pub fn establishment(&self) -> &'a ConnectionEstablishmentInfo {
         self.establishment
-283
+    }
-284
     /// Returns measurements from the successful establishment.
     /// Returns measurements frozen after protocol installation and before pool publication.
     pub fn stats(&self) -> &'a ConnectionEstablishmentStats {
         self.stats
-288
+    }
-289
     /// Returns immutable identity and transport facts for the connection.
     pub fn connection(&self) -> &'a PoolArc<ConnectionInfo> {
         self.connection
-293
+    }
-294
+}
-295

↧ Expand context

 /// Stable cause recorded when pool dispatch ownership ends.
 ///
 /// The later physical-close event may carry a more specific [`CloseReason`].
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[non_exhaustive]
 pub enum LogicalCloseCause {
     /// The connection exceeded its configured idle timeout.
     IdleTimeout,
     /// The connection was explicitly marked unsafe for reuse.
     Poisoned,
     /// Hyper's protocol dispatch ownership ended.
     ProtocolEnded,
     /// HTTP/1 did not prove a complete reusable message boundary.
     IncompleteH1Exchange,
     /// The connection closed to move bounded capacity to another cell.
     Reclaimed,
     /// The connection pool was dropped.
     PoolDropped,
     /// The runtime driving the connection shut down.
     OwnerRuntimeShutdown,

@@ -335,349 +611,655 @@

↥ Expand context

     /// Returns the first cause that ended pool dispatch ownership.
     pub fn cause(&self) -> LogicalCloseCause {
         self.cause
-352
+    }
-353
+}
-354
 /// Observation that the client released root transport ownership.
 ///
 /// This is the final event for an installed connection. It does not assert
 /// that peer, kernel, or TCP teardown has completed.
 #[derive(Debug)]
 pub struct ConnectionPhysicalClose<'a> {
     connection: &'a PoolArc<ConnectionInfo>,
     reason: CloseReason,
-363
+}
-364
 impl<'a> ConnectionPhysicalClose<'a> {
     /// Returns immutable identity and transport facts for the connection.
     pub fn connection(&self) -> &'a PoolArc<ConnectionInfo> {
         self.connection

-369
+    }
-370
     /// Returns the final protocol and ownership close classification.
     pub fn reason(&self) -> CloseReason {
         self.reason
-374
+    }
-375
+}
-376
 /// Pool-owned connection event source.
 ///
 /// This owner retains callback delivery, lifecycle timing, and establishment
 /// identity allocation. Starting an establishment performs no timing or
 /// identity work when no listener is configured.
 /// This owner retains callback delivery, establishment timing, and public
 /// establishment identity allocation. Timing is collected for successful
 /// connection metadata even when callback delivery is disabled.
 #[derive(Debug)]
 pub(super) struct ConnectionEvents {
     listener: Option<SharedConnectionEventListener>,
     time_source: SharedTimeSource,
     next_establishment_id: AtomicU64,
-387
+}
-388
 impl ConnectionEvents {
     /// Creates the event source retained by one connection pool.
     pub(super) fn new(
         listener: Option<SharedConnectionEventListener>,
         time_source: SharedTimeSource,
     ) -> Self {
         Self {
             listener,
             time_source,
             next_establishment_id: AtomicU64::new(0),
-399
+        }
-400
+    }
-401
     /// Records the start of an establishment when observation is enabled.
     /// Records the start of one connection establishment.
     pub(super) fn establishment_started(
         &self,
         origin: &OriginKey,
         partition: PartitionId,
         connection_stats: PoolArc<CellConnectionStats>,
     ) -> ConnectionEstablishment {
         connection_stats.establishment_started();
         let timing = EstablishmentTiming {
             time_source: self.time_source.clone(),
             started_at: self.time_source.now(),
             protocol_handshake_started_at: None,
             transport_duration: None,
             protocol_handshake_duration: None,
         };
         let observation = self
             .listener
             .as_ref()
             .map(|listener| EstablishmentObservation {
                 listener: listener.clone(),
                 info: ConnectionEstablishmentInfo {
                     id: ConnectionEstablishmentId(
                         self.next_establishment_id.fetch_add(1, Ordering::Relaxed),
                     ),
                     origin: origin.clone(),
                     partition,
                 },
                 time_source: self.time_source.clone(),
                 started_at: self.time_source.now(),
                 protocol_handshake_started_at: None,
                 transport_duration: None,
                 protocol_handshake_duration: None,
                 stage: ConnectionEstablishmentStage::Transport,
                 remote_addr: None,
                 protocol: None,
             });
         ConnectionEstablishment {
             timing,
             observation,
             successful_stats: None,
             connection_stats: Some(connection_stats),
-438
+        }
-439
+    }
-440
+}
-441
 /// Tracks observations collected during one connection establishment.
 /// Tracks measurements and optional callback state for one establishment.
 pub(super) struct ConnectionEstablishment {
     timing: EstablishmentTiming,
     observation: Option<EstablishmentObservation>,
     successful_stats: Option<ConnectionEstablishmentStats>,
     connection_stats: Option<PoolArc<CellConnectionStats>>,
-448
+}
-449
 /// Data collected while one observed establishment is active.
 struct EstablishmentObservation {
     listener: SharedConnectionEventListener,
     info: ConnectionEstablishmentInfo,
 /// Timing collected for every connection establishment.
 struct EstablishmentTiming {
     time_source: SharedTimeSource,
     started_at: SystemTime,
     protocol_handshake_started_at: Option<SystemTime>,
     transport_duration: Option<Duration>,
     protocol_handshake_duration: Option<Duration>,
-+
+}
-+
 /// Callback state retained only when a listener is configured.
 struct EstablishmentObservation {
     listener: SharedConnectionEventListener,
     info: ConnectionEstablishmentInfo,
     stage: ConnectionEstablishmentStage,
     remote_addr: Option<SocketAddr>,
     protocol: Option<ConnectionProtocol>,
-466
+}
-467
 impl ConnectionEstablishment {
     /// Records completion of the transport stage.
     pub(super) fn transport_completed(&mut self, remote_addr: Option<SocketAddr>) {
         self.timing.transport_duration = Some(self.timing.elapsed_since(self.timing.started_at));
         if let Some(observation) = &mut self.observation {
             observation.transport_duration =
                 Some(observation.elapsed_since(observation.started_at));
             observation.remote_addr = remote_addr;
             observation.stage = ConnectionEstablishmentStage::ProtocolSelection;
-475
+        }
-476
+    }
-477
     /// Records the protocol selected from connector metadata.
     pub(super) fn protocol_selected(&mut self, protocol: ConnectionProtocol) {
         if let Some(observation) = &mut self.observation {
             observation.protocol = Some(protocol);
-482
+        }
-483
+    }
-484
     /// Records entry into Hyper's protocol handshake.
     pub(super) fn protocol_handshake_started(&mut self) {
         self.timing.protocol_handshake_started_at = Some(self.timing.time_source.now());
         if let Some(observation) = &mut self.observation {
             observation.protocol_handshake_started_at = Some(observation.time_source.now());
             observation.stage = ConnectionEstablishmentStage::ProtocolHandshake;
-490
+        }
-491
+    }
-492
     /// Records a failed Hyper handshake before emitting its terminal event.
     pub(super) fn protocol_handshake_failed(&mut self) {
         if let Some(observation) = &mut self.observation {
             observation.finish_protocol_handshake();
--
+        }
         self.timing.finish_protocol_handshake();
-496
+    }
-497
     /// Records successful Hyper handshake before pool installation.
     pub(super) fn protocol_handshake_completed(&mut self) {
         self.timing.finish_protocol_handshake();
         if let Some(observation) = &mut self.observation {
             observation.finish_protocol_handshake();
             observation.stage = ConnectionEstablishmentStage::PoolInstallation;
-503
+        }
-504
+    }
-505
     /// Freezes successful measurements before the connection becomes visible.
     pub(super) fn installed(&mut self, connection: &PoolArc<ConnectionState>) {
         let stats = self.timing.stats();
         connection
             .info()
             .set_establishment(stats.connection_metadata());
         assert!(
             self.successful_stats.replace(stats).is_none(),
             "connection establishment was installed more than once"
         );
-+
+    }
-+
     /// Emits the terminal failure for this establishment.
     pub(super) fn failed(mut self, error: &ConnectorError) {
         let stats = self.timing.stats();
         self.finish_connection_stats();
         let Some(observation) = self.observation.take() else {
             return;
         };
         observation.notify_failure(error);
         observation.notify_failure(error, &stats);
-526
+    }
-527
     /// Reports successful installation and enables ordered close observations.
     pub(super) fn opened(mut self, connection: &PoolArc<ConnectionState>) {
         let stats = self.successful_stats.take().unwrap_or_else(|| {
             let stats = self.timing.stats();
             connection
                 .info()
                 .set_establishment(stats.connection_metadata());
             stats
         });
         self.finish_connection_stats();
         let listener = self.observation.take().map(|observation| {
             let stats = observation.stats();
             observation
                 .listener
                 .notify(&ConnectionEvent::Opened(ConnectionOpened {
                     establishment: &observation.info,
                     stats: &stats,
                     connection: connection.info(),
                 }));
             observation.listener
         });
         connection.complete_opened_event(listener.as_ref());
-549
+    }
-550
     /// Ends an establishment whose transport lost to existing HTTP/2 supply.
     pub(super) fn superseded(mut self) {
         self.finish_connection_stats();
         self.observation.take();
-555
+    }
-556
     fn finish_connection_stats(&mut self) {
         if let Some(stats) = self.connection_stats.take() {
             stats.establishment_finished();
-560
+        }
-561
+    }
-562
+}
-563
 impl Drop for ConnectionEstablishment {
     fn drop(&mut self) {
         let stats = self.successful_stats.unwrap_or_else(|| self.timing.stats());
         self.finish_connection_stats();
         let Some(observation) = self.observation.take() else {
             return;
         };
         let error = ConnectorError::io("connection establishment task was dropped".into());
         observation.notify_failure(&error);
         observation.notify_failure(&error, &stats);
-573
+    }
-574
+}
-575
 impl EstablishmentObservation {
     fn notify_failure(&self, error: &ConnectorError) {
         let stats = self.stats();
     fn notify_failure(&self, error: &ConnectorError, stats: &ConnectionEstablishmentStats) {
         self.listener.notify(&ConnectionEvent::EstablishmentFailed(
             ConnectionEstablishmentFailed {
                 establishment: &self.info,
                 stats: &stats,
                 stats,
                 stage: self.stage,
                 remote_addr: self.remote_addr,
                 protocol: self.protocol,
                 error,
             },
         ));
-588
+    }
-+
+}
-590
 impl EstablishmentTiming {
     fn stats(&self) -> ConnectionEstablishmentStats {
         let total_duration = self.elapsed_since(self.started_at);
         ConnectionEstablishmentStats {
             total_duration,
             transport_duration: self.transport_duration.unwrap_or(total_duration),
             protocol_handshake_duration: self.protocol_handshake_duration.or_else(|| {
                 self.protocol_handshake_started_at
                     .map(|started_at| self.elapsed_since(started_at))
             }),
-601
+        }
-602
+    }
-603
     fn finish_protocol_handshake(&mut self) {
         self.protocol_handshake_duration = self
             .protocol_handshake_started_at
             .take()
             .map(|started_at| self.elapsed_since(started_at));
-609
+    }
-610
     fn elapsed_since(&self, started_at: SystemTime) -> Duration {
         self.time_source
             .now()
             .duration_since(started_at)
             .unwrap_or_default()
-616
+    }
-617
+}
-618
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
     use aws_smithy_async::test_util::ManualTimeSource;
     use std::sync::atomic::AtomicUsize;
     use std::sync::Mutex;
     use std::time::UNIX_EPOCH;
-626
     fn origin() -> OriginKey {
         OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "example.com", None).unwrap()
-629
+    }
-630
     fn events(listener: Option<SharedConnectionEventListener>) -> ConnectionEvents {
         ConnectionEvents::new(listener, SharedTimeSource::default())
-633
+    }
-634
     fn establishment(events: &ConnectionEvents) -> ConnectionEstablishment {

↧ Expand context

         events.establishment_started(
             &origin(),
             PartitionId::from_index(1),
             PoolArc::new(CellConnectionStats::default()),
-640
+        )
-641
+    }
-642
     #[test]
     fn disabled_callbacks_do_not_allocate_establishment_ids() {
         let events = events(None);
         events.next_establishment_id.store(7, Ordering::Relaxed);
-647
         establishment(&events).superseded();
-649
         assert_eq!(7, events.next_establishment_id.load(Ordering::Relaxed));
-651
+    }
-652
     #[test]
     fn listener_panic_is_isolated() {
         let events = events(Some(SharedConnectionEventListener::new(

@@ -675,719 +734,821 @@

↥ Expand context

                     panic!("unexpected event: {event:?}");
                 };
                 *observed.lock().unwrap() = Some((
                     failed.establishment().id(),
                     failed.stage(),
                     failed.protocol(),
                     failed.stats().protocol_handshake_duration().is_some(),
                 ));
-727
+            }
         })));
         let mut establishment = establishment(&events);
         establishment.transport_completed(None);
         establishment.protocol_selected(ConnectionProtocol::Http2);
         establishment.protocol_handshake_started();
         establishment.protocol_handshake_failed();
         let error = ConnectorError::io("synthetic handshake failure".into());
-735
         establishment.failed(&error);
-737
         assert_eq!(

             Some((
                 ConnectionEstablishmentId(0),
                 ConnectionEstablishmentStage::ProtocolHandshake,
                 Some(ConnectionProtocol::Http2),
                 true,
             )),
             *observed.lock().unwrap()
         );
-747
+    }
-748
     #[test]
     fn establishment_stats_measure_recorded_phases() {
         let time = ManualTimeSource::new(UNIX_EPOCH);
         let observed = Arc::new(Mutex::new(None));
         let events = ConnectionEvents::new(
             Some(SharedConnectionEventListener::new({
                 let observed = observed.clone();
                 move |event: &ConnectionEvent<'_>| {
                     let ConnectionEvent::EstablishmentFailed(failed) = event else {
                         panic!("unexpected event: {event:?}");
                     };
                     *observed.lock().unwrap() = Some(*failed.stats());
-+
+                }
             })),
             SharedTimeSource::new(time.clone()),
         );
         let mut establishment = establishment(&events);
-+
         time.advance(Duration::from_secs(2));
         establishment.transport_completed(None);
         establishment.protocol_selected(ConnectionProtocol::Http2);
         establishment.protocol_handshake_started();
         time.advance(Duration::from_secs(3));
         establishment.protocol_handshake_completed();
         time.advance(Duration::from_secs(2));
         establishment.failed(&ConnectorError::io("synthetic installation failure".into()));
-+
         let stats = observed.lock().unwrap().expect("establishment stats");
         assert_eq!(stats.total_duration(), Duration::from_secs(7));
         assert_eq!(stats.transport_duration(), Duration::from_secs(2));
         assert_eq!(
             stats.protocol_handshake_duration(),
             Some(Duration::from_secs(3))
         );
         let metadata = stats.connection_metadata();
         assert_eq!(metadata.total_duration(), Duration::from_secs(7));
         assert_eq!(metadata.transport_duration(), Duration::from_secs(2));
         assert_eq!(
             metadata.protocol_handshake_duration(),
             Some(Duration::from_secs(3))
         );
-+
+    }
-+
     #[test]
     fn superseded_establishment_emits_no_public_event() {
         let observed = Arc::new(AtomicUsize::new(0));
         let events = events(Some(SharedConnectionEventListener::new({
             let observed = observed.clone();
             move |_: &ConnectionEvent<'_>| {
                 observed.fetch_add(1, Ordering::Relaxed);
-799
+            }
         })));
-801

↧ Expand context

         establishment(&events).superseded();
-803
         assert_eq!(0, observed.load(Ordering::Relaxed));
         assert_eq!(1, events.next_establishment_id.load(Ordering::Relaxed));
-806
+    }
-807
     #[test]
     fn closure_listener_receives_one_terminal_event() {
         let observed = Arc::new(AtomicUsize::new(0));
         let events = events(Some(SharedConnectionEventListener::new({
             let observed = observed.clone();
             move |event: &ConnectionEvent<'_>| {
                 assert!(matches!(event, ConnectionEvent::EstablishmentFailed(_)));
                 observed.fetch_add(1, Ordering::Relaxed);
-816
+            }
         })));
         let error = ConnectorError::io("synthetic transport failure".into());
-819
         establishment(&events).failed(&error);
-821

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

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

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

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

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

↧ Expand context

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

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

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

↥ Expand context

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

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

↧ Expand context

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

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

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

                             connection_id,
                             stream_id,
                             method,
                             path: path.clone(),
                         });
                         let Some(stream_script) = script.script_for(&path) else {
                             respond.send_reset(Reason::PROTOCOL_ERROR);
                             return Err(H2HarnessError::new(format!(
                                 "connection {connection_id:?} received unscripted H2 path {path:?}"
                             )));
                         };
                         let held_request = match &stream_script {
                             H2StreamScript::RespondBeforeReceivingRequestBody { receive, .. } => {
                                 Some((request.into_body(), receive.clone()))
-850
+                            }
                             _ => {
                                 drop(request);
                                 None
-854
+                            }
                         };

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

↥ Expand context

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

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

↧ Expand context

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

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

@@ -0,1 +0,133 @@

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

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

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

↥ Expand context

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

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

↧ Expand context

         origin: OriginKey,
         stage: ConnectionEstablishmentStage,
         remote_addr: Option<SocketAddr>,
         protocol: Option<ConnectionProtocol>,
         error: String,
     },
     Opened {
         establishment: ConnectionEstablishmentId,
         connection: aws_smithy_runtime_api::client::connection::ConnectionId,
         establishment_partition: PartitionId,
         establishment_origin: OriginKey,
         connection_origin: OriginKey,
         connection_partition: PartitionId,
         remote_addr: Option<SocketAddr>,
         protocol: ConnectionProtocol,
         connect_path: ConnectPath,
         handshake_measured: bool,
     },
-70
+}
-71

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

↥ Expand context

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

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

↧ Expand context

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

@@ -410,585 +469,734 @@

↥ Expand context

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

     assert_eq!(1, harness.tcp_accepted_count());
-606
     drop(first_connector);
     drop(second_connector);
     drop(first_client);
     drop(second_client);
     drop(pool);
     harness.shutdown().await.expect("clean harness shutdown");
-613
+}
-614
 #[tokio::test]
 async fn peer_h1_reuse_survives_borrower_runtime_shutdown() {
     use runtime::DrivenRuntime;
-+
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             Http1Script::responses([
                 Http1Response::ok().body("owner"),
                 Http1Response::ok().body("peer"),
                 Http1Response::ok().body("reused"),
             ]),
-+
+        )
         .build()
         .await
         .expect("harness should start");
     let first_id = PartitionId::from_index(1);
     let second_id = PartitionId::from_index(2);
     let first_runtime = DrivenRuntime::start("h1-owner-one");
     let second_runtime = DrivenRuntime::start("h1-owner-two");
     assert_ne!(first_runtime.id(), second_runtime.id());
     let pool = ConnectionPool::builder()
         .partitions([
             Partition::new(first_id, first_runtime.driver_spawner()),
             Partition::new(second_id, second_runtime.driver_spawner()),
         ])
         .connection_reuse_scope(ConnectionReuseScope::Pool)
         .max_connections_per_host(1)
         .build_http()
         .expect("valid pool");
     let first_client = shared_client(&pool, first_id);
     let second_client = shared_client(&pool, second_id);
     let first_connector = connector(&first_client);
     let second_connector = connector(&second_client);
     let url = harness.endpoint_url();
-+
     let first_request = first_connector.clone();
     let first_url = url.clone();
     let (status, body) = first_runtime
         .spawn(async move { test_client::get_and_collect(&first_request, &first_url).await })
         .await
         .expect("owner request task should not panic");
     assert_eq!((status, body.as_slice()), (200, b"owner".as_slice()));
-+
     let peer_request = second_connector.clone();
     let peer_url = url.clone();
     let (status, body) = second_runtime
         .spawn(async move { test_client::get_and_collect(&peer_request, &peer_url).await })
         .await
         .expect("peer request task should not panic");
     assert_eq!((status, body.as_slice()), (200, b"peer".as_slice()));
     assert!(first_runtime.submitted_tasks() > 0);
     assert!(second_runtime.submitted_tasks() > 0);
-+
     second_runtime.shutdown();
-+
     let reused_request = first_connector.clone();
     let reused_url = url.clone();
     let (status, body) = first_runtime
         .spawn(async move { test_client::get_and_collect(&reused_request, &reused_url).await })
         .await
         .expect("reused request task should not panic");
     assert_eq!((status, body.as_slice()), (200, b"reused".as_slice()));
-+
     let request_connections = harness
         .events()
         .into_iter()
         .filter_map(|event| match event {
             WireConnectionEvent::Http1Request { connection_id, .. } => Some(connection_id),
             _ => None,
         })
         .collect::<Vec<_>>();
     assert_eq!(3, request_connections.len());
     assert!(
         request_connections
             .iter()
             .all(|connection| *connection == request_connections[0]),
         "owner and peer requests should use one HTTP/1 connection"
     );
     assert_eq!(1, harness.tcp_accepted_count());
-+
     drop(first_connector);
     drop(second_connector);
     drop(first_client);
     drop(second_client);
     drop(pool);
     first_runtime.shutdown();
     harness.shutdown().await.expect("clean harness shutdown");
-+
+}
-+
 #[tokio::test]
 async fn ineligible_partition_reclaims_peer_capacity() {
     let harness = ConnectionTestHarness::builder()
         .endpoint(
             IP1,
             EndpointPlan::queue([
                 Http1Script::responses([Http1Response::ok().body("first")]),
                 Http1Script::responses([Http1Response::ok().body("second")]),
             ]),
-714
+        )

↧ Expand context

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

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

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

↥ Expand context

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

Pages: 1 2 3 4

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