AWS SDK

rev. 39ed07918eb1729475c85ddf93a83d0b1539e928 (ignoring whitespace)

Files changed:

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

@@ -1,1 +40,41 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Partition-bound handles for Smithy HTTP operations.
-7
 use super::partition::PartitionId;
 use super::registry::PartitionState;
 use super::ConnectionPool;
 use crate::client::downcast_error;
 use crate::client::timeout::{self, TimeoutKind};
 use crate::sync::Arc;
 use aws_smithy_async::rt::sleep::{default_async_sleep, SharedAsyncSleep};
 use aws_smithy_runtime_api::client::connector_metadata::ConnectorMetadata;
 use aws_smithy_runtime_api::client::http::{
     HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpConnector,
 };
 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;

↧ Expand context

 use std::borrow::Cow;
 use std::error::Error;
 use std::fmt;
 use std::time::Duration;
-26
 /// Smithy HTTP client bound to one connection-pool partition.
 ///
 /// Construction resolves the partition once. Every request uses that partition
 /// for establishment placement and local reuse, while the configured reuse
 /// scope may permit dispatch through a peer partition's connection. Cloning a
 /// client shares the pool, resolved partition state, and reusable connections.
 #[derive(Clone)]
 pub struct Client {
     /// Shared connection pool and immutable policy.
     pool: ConnectionPool,
     /// Resolved runtime and network placement for this client.
     partition: Arc<PartitionState>,
-39
+}
-40
 impl Client {

@@ -171,172 +245,244 @@

↥ Expand context

     read_timeout: Option<Duration>,
     /// Runtime timer used by operation timeouts.
     sleep: Option<SharedAsyncSleep>,
-175
+}
-176
 impl fmt::Debug for PoolConnector {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("PoolConnector")
             .field("partition", &self.partition.id())
             .field("connect_timeout", &self.connect_timeout)
             .field("read_timeout", &self.read_timeout)
             .finish_non_exhaustive()
-184
+    }
-185
+}
-186
 impl HttpConnector for PoolConnector {
     fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
         let pool = self.pool.clone();
         let partition = self.partition.clone();
         let connect_timeout = self.connect_timeout;

         let read_timeout = self.read_timeout;
         let sleep = self.sleep.clone();
         HttpConnectorFuture::new(async move {
             if (connect_timeout.is_some() || read_timeout.is_some()) && sleep.is_none() {
                 return Err(ConnectorError::user(MissingAsyncSleep.into()));
-197
+            }
-198
             let request = request.try_into_http1x().map_err(|error| {
                 aws_smithy_runtime_api::client::result::ConnectorError::user(error.into())
             })?;
             let send = pool.send_request(
                 partition,
                 request,
                 connect_timeout.zip(sleep.clone()).map(|(duration, sleep)| {
                     super::establish::TransportTimeout::new(duration, sleep)
                 }),
             );
             let options =
                 super::dispatch::RequestOptions::new(connect_timeout.zip(sleep.clone()).map(
                     |(duration, sleep)| super::establish::TransportTimeout::new(duration, sleep),
                 ));
             let send = pool.send_request(partition, request, options);
             let response = timeout::maybe_timeout_future(
                 send,
                 read_timeout,
                 sleep.as_ref(),
                 TimeoutKind::Read,
-212
+            )
             .await
             .map_err(super::super::downcast_error)?;
             .map_err(downcast_error)?;
             HttpResponse::try_from(response).map_err(|error| {
                 aws_smithy_runtime_api::client::result::ConnectorError::other(error.into(), None)
             })
         })
-219
+    }
-220
+}
-221
 /// Operation timeouts were configured without a runtime timer.
 #[derive(Debug)]
 struct MissingAsyncSleep;

↧ Expand context

-225
 impl fmt::Display for MissingAsyncSleep {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str("an async sleep implementation is required for HTTP connect or read timeouts")
-229
+    }
-230
+}
-231
 impl Error for MissingAsyncSleep {}
-233
 #[cfg(all(test, feature = "rt-tokio", not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
-237
     #[tokio::test]
     async fn configured_timeout_without_sleep_returns_a_user_error() {
         let pool = ConnectionPool::builder()
             .idle_timeout(None)
             .build_http()
             .unwrap();
         let client = Client::new(&pool).unwrap();

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

@@ -1,1 +44,43 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Protocol-independent connection lifetime.
 //!
 //! [`ConnectionState`] serializes dispatch commitment with logical close.
 //! [`DispatchGuard`] accounts for accepted requests.
 //! [`PhysicalConnectionGuard`] follows root I/O until the pool no longer owns
 //! it, including transfer through a protocol upgrade. This physical transition
 //! does not assert that the peer or kernel has completed the underlying TCP
 //! close. Logical close returns bounded capacity without waiting for either
 //! lifetime to finish.
 //! [`PhysicalConnectionGuard`] follows root I/O until the client releases its
 //! physical connection handle, including transfer through a protocol upgrade.
 //! The operating system may continue TCP teardown afterward. Logical close
 //! returns bounded capacity without waiting for either lifetime to finish.
-14
 use super::admission::CapacityLease;
 use super::origin::OriginKey;
 use super::partition::PartitionId;
 use crate::sync::{Arc, Mutex};
 pub use aws_smithy_runtime_api::client::connection::ConnectionId;
 use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
 use http_1x::Extensions;
 use hyper::rt::{Read, ReadBufCursor, Write};
 use hyper_util::client::legacy::connect::{Connected, Connection, HttpInfo};

↧ Expand context

 use pin_project_lite::pin_project;
 use std::fmt;
 use std::io::{self, IoSlice};
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-30
 /// Why a connection stopped accepting new work.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[non_exhaustive]
 pub enum CloseReason {
     /// The connection exceeded its configured idle timeout.
     IdleTimeout,
     /// The connection was explicitly marked unsafe for reuse.
     Poisoned,
     /// The protocol driver or peer closed the connection.
     ProtocolClosed,
     /// HTTP/1 did not prove a complete reusable message boundary.
     IncompleteH1Exchange,
     /// The transport left HTTP/1 pool ownership through an upgrade.

@@ -170,169 +321,320 @@

↥ Expand context

         let mut builder = ConnectionMetadata::builder()
             .proxied(self.proxied)
             .connection_id(self.id)
             .poison_fn(move || {
                 close.close(CloseReason::Poisoned);
             });
         builder
             .set_local_addr(self.local_addr)
             .set_remote_addr(self.remote_addr);
         builder.build()
-179
+    }
-180
     /// Creates synthetic HTTP/1 information for state-machine tests.
     #[cfg(test)]
     pub(super) fn for_test(id: ConnectionId, owner_partition: PartitionId) -> Arc<Self> {
         Self::new(
             id,
             OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "example.com", None)
                 .expect("synthetic test origin is valid"),
             owner_partition,

             NegotiatedProtocol::Http1,
             Connected::new(),
-191
+        )
-192
+    }
-193
+}
-194
 /// Shared connection ownership from negotiated transport through close.
 pub(super) struct ConnectionState {
     /// Identity and transport facts shared with metadata and lifecycle events.
     info: Arc<ConnectionInfo>,
     /// Dispatch, logical-close, and root-I/O completion state.
     lifecycle: Mutex<LifecycleState>,
     /// Dispatch, logical-close, and physical-connection completion state.
     lifecycle: Mutex<ConnectionLifecycle>,
-201
+}
-202
 /// Connection lifetime state serialized with dispatch commitment and close.
 #[derive(Debug)]
 struct LifecycleState {
 struct ConnectionLifecycle {
     /// Dispatch eligibility and ownership of bounded capacity.
     logical: LogicalState,
     /// Requests that committed before logical close.
     in_flight: usize,
     /// Whether ownership of root transport I/O has ended.
     physical_complete: bool,
     /// Whether the client has released its physical connection handle.
     physical_connection_complete: bool,
-212
+}
-213
 /// Whether a connection may accept dispatch and still owns bounded capacity.
 #[derive(Debug)]
 enum LogicalState {
     /// The connector returned connected I/O and selected the HTTP protocol,
     /// but Hyper has not produced the request handle required for dispatch.
     PendingOpen,
     /// Dispatch may commit.
     Open {
         /// Bounded-origin slot released by logical close, when configured.
         lease: Option<CapacityLease>,
         capacity: Option<CapacityLease>,
     },
     /// New dispatch is rejected while accepted work may still drain.
     Closed {
         /// Reason recorded by the first logical-close transition.
         reason: CloseReason,
     },
-230
+}
-231
 impl ConnectionState {
     /// Creates state after transport establishment and protocol selection.
     ///
     /// For TLS transports, the connector has completed TLS and ALPN before this
     /// call. Hyper protocol setup and cell installation have not occurred.
     ///
     /// The returned guard is the unique physical-lifetime owner. After Hyper
     /// returns a request handle, [`Self::open`] attaches optional bounded
     /// capacity before cell installation makes the connection discoverable.
     /// The returned guard uniquely tracks the client's physical connection
     /// handle.
     /// After Hyper returns a request handle, [`Self::open`] attaches optional
     /// bounded capacity before cell installation makes the connection
     /// discoverable.
     pub(super) fn pending_open(info: Arc<ConnectionInfo>) -> (Arc<Self>, PhysicalConnectionGuard) {
         let connection = Arc::new(Self {
             info,
             lifecycle: Mutex::new(LifecycleState {
             lifecycle: Mutex::new(ConnectionLifecycle {
                 logical: LogicalState::PendingOpen,
                 in_flight: 0,
                 physical_complete: false,
                 physical_connection_complete: false,
             }),
         });
         let physical = PhysicalConnectionGuard {
             connection: connection.clone(),
             active: true,
         };
         (connection, physical)
-257
+    }
-258
     /// Opens dispatch commitment and transfers optional bounded capacity.
     ///
     /// The caller performs this transition after Hyper protocol setup and
     /// before publishing the connection through its cell. Opening first
     /// ensures that a newly visible request handle can commit dispatch.
     ///
     /// Returns `lease` when logical close won before opening.
     pub(super) fn open(&self, lease: Option<CapacityLease>) -> Result<(), Option<CapacityLease>> {
         let mut lifecycle = self.lifecycle.lock();
         if !matches!(lifecycle.logical, LogicalState::PendingOpen) {
             return Err(lease);
-270
+        }
         lifecycle.logical = LogicalState::Open { lease };
         lifecycle.logical = LogicalState::Open { capacity: lease };
         Ok(())
-273
+    }
-274
     /// Creates a connection whose origin has no admission bound.
     ///
     /// The returned guard is the unique physical-lifetime owner and must move
     /// with the root I/O task.
     /// The returned unique guard must move with the root I/O task.
     #[cfg(test)]
     pub(super) fn unbounded(info: Arc<ConnectionInfo>) -> (Arc<Self>, PhysicalConnectionGuard) {
         let (connection, physical) = Self::pending_open(info);
         connection
             .open(None)
             .expect("new unbounded connection could not open");
         (connection, physical)
-285
+    }
-286
     /// Creates a connection that takes ownership of one bounded-origin slot.
     ///
     /// The returned guard is the unique physical-lifetime owner and must move
     /// with the root I/O task. Logical close returns `lease` independently of
     /// that guard.
     /// The returned unique guard must move with the root I/O task. Logical
     /// close returns `lease` independently of that guard.
     #[cfg(test)]
     pub(super) fn bounded(
         info: Arc<ConnectionInfo>,
         lease: CapacityLease,
     ) -> (Arc<Self>, PhysicalConnectionGuard) {
         let (connection, physical) = Self::pending_open(info);
         connection
             .open(Some(lease))
             .expect("new bounded connection could not open");
         (connection, physical)

↧ Expand context

-301
+    }
-302
     /// Returns this connection's stable identity.
     pub(super) fn id(&self) -> ConnectionId {
         self.info.id()
-306
+    }
-307
     /// Returns the partition that owns this connection's I/O and driver.
     pub(super) fn owner_partition(&self) -> PartitionId {
         self.info.owner_partition()
-311
+    }
-312
     /// Returns immutable identity and transport facts for this connection.
     pub(super) fn info(&self) -> &Arc<ConnectionInfo> {
         &self.info
-316
+    }
-317
     /// Attempts to commit one request against logical close.
     ///
     /// Returns a guard and increments the in-flight count while the connection

@@ -326,325 +612,611 @@

↥ Expand context

             return None;
-326
+        }
         lifecycle.in_flight = lifecycle
             .in_flight
             .checked_add(1)
             .expect("in-flight dispatch count exhausted");
         drop(lifecycle);
-332
         Some(DispatchGuard {
             connection: connection.clone(),
             active: true,
         })
-337
+    }
-338
     /// Performs the first logical-close transition.
     ///
     /// Returns `true` when this call closes the connection and records
     /// `reason`. Returns `false` when another close already won; the original
     /// reason remains unchanged.
     ///

     /// The detached lease is dropped only after the connection lock is
     /// released, so admission and connection locks are never nested.
     pub(super) fn logical_close(&self, reason: CloseReason) -> bool {
         let lease = {
             let mut lifecycle = self.lifecycle.lock();
             match &mut lifecycle.logical {
                 LogicalState::PendingOpen => {
                     lifecycle.logical = LogicalState::Closed { reason };
                     None
-354
+                }
                 LogicalState::Open { lease } => {
                     let lease = lease.take();
                 LogicalState::Open { capacity } => {
                     let lease = capacity.take();
                     lifecycle.logical = LogicalState::Closed { reason };
                     lease
-359
+                }
                 LogicalState::Closed { .. } => return false,
-361
+            }
         };
         drop(lease);
         tracing::debug!(
             connection_id = %self.id(),
             connection_partition = ?self.owner_partition(),
             protocol = ?self.info.protocol(),
             origin_scheme = %self.info.origin().scheme(),
             origin_host = self.info.origin().host(),
             origin_port = ?self.info.origin().port(),
             close_reason = ?reason,
             "connection logically closed"
         );
         true
-375
+    }
-376
     /// Refines a driver-observed close after Hyper confirms an upgrade.
     ///
     /// Hyper may transfer upgraded I/O and complete its HTTP/1 driver in one
     /// poll. The driver can therefore record [`CloseReason::ProtocolClosed`]
     /// before the request task observes the upgrading response. This method
     /// changes that stored reason to [`CloseReason::Upgraded`].
     ///
     /// Returns `true` only when that refinement was applied. It does not close
     /// the connection again, change I/O ownership, or release capacity a second
     /// time.
     pub(super) fn refine_protocol_close_as_upgrade(&self) -> bool {
         let refined = {
             let mut lifecycle = self.lifecycle.lock();
             match &mut lifecycle.logical {
                 LogicalState::Closed { reason } if *reason == CloseReason::ProtocolClosed => {
                     *reason = CloseReason::Upgraded;
                     true
-394
+                }
                 _ => false,
-396
+            }
         };
         if refined {
             tracing::debug!(
                 connection_id = %self.id(),
                 connection_partition = ?self.owner_partition(),
                 origin_scheme = %self.info.origin().scheme(),
                 origin_host = self.info.origin().host(),
                 origin_port = ?self.info.origin().port(),
                 previous_close_reason = ?CloseReason::ProtocolClosed,
                 close_reason = ?CloseReason::Upgraded,
                 "refined connection close reason after HTTP/1 upgrade"
             );
-409
+        }
         refined
-411
+    }
-412
     /// Verifies a close reason at a protocol handoff boundary in debug builds.
     #[cfg(debug_assertions)]
     /// Verifies a close reason at a protocol handoff boundary in debug builds and tests.
     #[cfg(any(debug_assertions, test))]
     pub(super) fn debug_assert_close_reason(&self, expected: CloseReason) {
         let lifecycle = self.lifecycle.lock();
         let actual = match lifecycle.logical {
             LogicalState::PendingOpen => None,
             LogicalState::Open { .. } => None,
             LogicalState::Closed { reason } => Some(reason),
         };
         debug_assert_eq!(Some(expected), actual);
-423
+    }
-424
     /// Removes one dispatch previously committed by [`Self::try_commit_dispatch`].
     fn finish_dispatch(&self) {
     fn release_dispatch(&self) {
         let mut lifecycle = self.lifecycle.lock();
         lifecycle.in_flight = lifecycle
             .in_flight
             .checked_sub(1)
             .expect("completed a dispatch that was not in flight");
-432
+    }
-433
     /// Records that the connection's root I/O is no longer live.
     /// Records that the client released its physical connection handle.
     ///
     /// # Panics
     ///
     /// Panics if physical ownership completes more than once.
     fn finish_physical(&self) {
     /// Panics if physical connection ownership completes more than once.
     fn complete_physical_connection(&self) {
-440
+        {
             let mut lifecycle = self.lifecycle.lock();
             assert!(
                 !lifecycle.physical_complete,
                 !lifecycle.physical_connection_complete,
                 "physical connection ownership completed more than once"
             );
             lifecycle.physical_complete = true;
             lifecycle.physical_connection_complete = true;
-447
+        }
         tracing::debug!(
             connection_id = %self.id(),
             connection_partition = ?self.owner_partition(),
             origin_scheme = %self.info.origin().scheme(),
             origin_host = self.info.origin().host(),
             origin_port = ?self.info.origin().port(),
             "connection root I/O ownership ended"
             "physical connection ownership ended"
         );
-456
+    }
-457
     /// Returns a consistent lifecycle snapshot.
     #[cfg(test)]
     pub(super) fn snapshot(&self) -> ConnectionSnapshot {
     pub(super) fn probe(&self) -> ConnectionProbe {
         let lifecycle = self.lifecycle.lock();
         ConnectionSnapshot {
         ConnectionProbe {
             close_reason: match lifecycle.logical {
                 LogicalState::PendingOpen => None,
                 LogicalState::Open { .. } => None,
                 LogicalState::Closed { reason } => Some(reason),
             },
             in_flight: lifecycle.in_flight,
             physical_complete: lifecycle.physical_complete,
             physical_connection_complete: lifecycle.physical_connection_complete,
-470
+        }
-471
+    }
-472
+}
-473
 impl fmt::Debug for ConnectionState {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("ConnectionState")
             .field("info", &self.info)
             .field("lifecycle", &self.lifecycle)
             .finish()
-480
+    }
-481
+}
-482
 #[cfg(test)]
 /// Observable lifecycle state used by protocol coordination and focused tests.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) struct ConnectionSnapshot {
 pub(super) struct ConnectionProbe {
     /// Reason logical close won, or `None` while dispatch is accepted.
     pub(super) close_reason: Option<CloseReason>,
     /// Number of dispatches that have not completed.
     pub(super) in_flight: usize,
     /// Whether root-I/O ownership has completed.
     pub(super) physical_complete: bool,
     /// Whether the client released its physical connection handle.
     pub(super) physical_connection_complete: bool,
-493
+}
-494
 /// One dispatch that committed before logical close.
 ///
 /// Dropping the guard records request completion exactly once.
 #[derive(Debug)]
 pub(super) struct DispatchGuard {
     /// Shared state whose in-flight count this guard owns.
     connection: Arc<ConnectionState>,
     /// Whether `Drop` still owes request completion.
     active: bool,
-504
+}
-505
 impl DispatchGuard {
     /// Returns the identity of the connection carrying this dispatch.
     #[cfg(test)]
     pub(super) fn connection_id(&self) -> ConnectionId {
         self.connection.id()
-511
+    }
-512
     /// Consumes the guard and records request completion immediately.
     ///
     /// Dropping an uncompleted guard performs the same accounting as a
     /// cancellation fallback.
     pub(super) fn complete(mut self) {
     pub(super) fn release(mut self) {
         self.active = false;
         self.connection.finish_dispatch();
         self.connection.release_dispatch();
-520
+    }
-521
+}
-522
 impl Drop for DispatchGuard {
     fn drop(&mut self) {
         if self.active {
             self.connection.finish_dispatch();
             self.connection.release_dispatch();
-527
+        }
-528
+    }
-529
+}
-530
 /// Unique root-I/O ownership whose drop records the end of pool ownership.
 /// Tracks the lifetime of the client's physical connection handle.
 ///
 /// The guard is created with the connection and moves with root I/O through
 /// protocol drain or upgrade. Completion says only that the pool no longer
 /// owns the root transport; the operating system may continue TCP teardown.
 /// protocol drain or upgrade. Dropping it means the client no longer owns its
 /// transport handle; the operating system may continue TCP teardown.
 #[derive(Debug)]
 pub(super) struct PhysicalConnectionGuard {
     /// Shared state whose physical lifetime this guard owns.
     /// Shared state whose physical connection lifetime this guard tracks.
     connection: Arc<ConnectionState>,
     /// Whether `Drop` still owes physical completion.
     /// Whether `Drop` still owes physical connection completion.
     active: bool,
-542
+}
-543
 impl PhysicalConnectionGuard {
     /// Consumes the guard and records root-I/O completion immediately.
     /// Consumes the guard and records physical connection completion.
     ///
     /// Dropping an uncompleted guard performs the same transition during task
     /// cancellation or runtime shutdown.
     #[cfg(test)]
     pub(super) fn complete(mut self) {
     pub(super) fn release(mut self) {
         self.active = false;
         self.connection.finish_physical();
         self.connection.complete_physical_connection();
-553
+    }
-554
+}
-555
 impl Drop for PhysicalConnectionGuard {
     fn drop(&mut self) {
         if self.active {
             self.connection.finish_physical();
             self.connection.complete_physical_connection();
-560
+        }
-561
+    }
-562
+}
-563
 pin_project! {
     /// Root transport wrapper that tracks how long the pool owns the I/O.
     /// Root transport wrapper that ties physical connection ownership to I/O.
     ///
     /// The wrapper moves intact through Hyper's driver and H1 upgrade path.
     /// Logical close may happen earlier. Dropping the wrapper records that pool
     /// ownership ended after the wrapped I/O was destroyed; it does not observe
     /// peer or kernel TCP completion.
     /// Logical close may happen earlier. Dropping the wrapper records that the
     /// client released its physical connection handle after the wrapped I/O
     /// was destroyed; the operating system may continue TCP teardown.
     pub(super) struct ConnectionIo<T> {
         #[pin]
         inner: T,
         // Declared after `inner` so transport destruction precedes the
         // physical-completion signal.
         // physical-connection completion signal.
         physical: PhysicalConnectionGuard,
-577
+    }
-578
+}
-579
 impl<T> ConnectionIo<T> {
     /// Attaches the unique physical-lifetime guard to root transport I/O.
     /// Attaches physical connection lifetime tracking to root transport I/O.
     pub(super) fn new(inner: T, physical: PhysicalConnectionGuard) -> Self {
         Self { inner, physical }
-584
+    }
-585
     /// Returns the wrapped transport.
     #[cfg(test)]
     pub(super) fn get_ref(&self) -> &T {
         &self.inner
-590
+    }
-591
+}

↧ Expand context

-592
 impl<T> fmt::Debug for ConnectionIo<T>
 where
     T: fmt::Debug,
-596
+{
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("ConnectionIo")
             .field("inner", &self.inner)
             .finish_non_exhaustive()
-601
+    }
-602
+}
-603
 impl<T> Connection for ConnectionIo<T>
 where
     T: Connection,
-607
+{
     fn connected(&self) -> Connected {
         self.inner.connected()
-610
+    }
-611
+}

@@ -641,640 +848,841 @@

↥ Expand context

-640
+    }
-641
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         Write::poll_shutdown(self.project().inner, cx)
-644
+    }
-645
     fn is_write_vectored(&self) -> bool {
         self.inner.is_write_vectored()
-648
+    }
-649
     fn poll_write_vectored(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         bufs: &[IoSlice<'_>],
     ) -> Poll<io::Result<usize>> {
         Write::poll_write_vectored(self.project().inner, cx, bufs)
-656
+    }
-657
+}
-658
 #[cfg(all(test, not(smithy_http_client_loom)))]

 mod tests {
     use super::*;
     use crate::client::pool::admission::OriginAdmission;
     use std::num::NonZeroUsize;
-664
     fn test_info(id: u64) -> Arc<ConnectionInfo> {
         ConnectionInfo::for_test(ConnectionId::new(id), PartitionId::from_index(0))
-667
+    }
-668
     #[test]
     fn logical_close_releases_capacity_before_physical_completion() {
     fn logical_close_releases_capacity_before_physical_connection_completion() {
         let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
         let lease = OriginAdmission::lease_for_test(&origin);
         let (connection, physical) = ConnectionState::bounded(test_info(1), lease);
-674
         assert!(connection.logical_close(CloseReason::Reclaimed));
         assert!(!connection.logical_close(CloseReason::PoolDropped));
         assert_eq!(1, origin.available_capacity_for_test());
         assert!(!connection.snapshot().physical_complete);
         assert!(!connection.probe().physical_connection_complete);
         assert_eq!(
             Some(CloseReason::Reclaimed),
             connection.snapshot().close_reason
             connection.probe().close_reason
         );
-683
         drop(physical);
         assert!(connection.snapshot().physical_complete);
         assert!(connection.probe().physical_connection_complete);
-686
+    }
-687
     #[test]
     fn upgrade_refines_only_a_driver_observed_protocol_close() {
         let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
         let lease = OriginAdmission::lease_for_test(&origin);
         let (connection, _physical) = ConnectionState::bounded(test_info(1), lease);
-693
         assert!(connection.logical_close(CloseReason::ProtocolClosed));
         assert!(connection.refine_protocol_close_as_upgrade());
         assert_eq!(
             Some(CloseReason::Upgraded),
             connection.snapshot().close_reason
         );
         assert_eq!(Some(CloseReason::Upgraded), connection.probe().close_reason);
         assert_eq!(1, origin.available_capacity_for_test());
         assert!(!connection.refine_protocol_close_as_upgrade());
-699
         let (poisoned, _physical) = ConnectionState::unbounded(test_info(2));
         assert!(poisoned.logical_close(CloseReason::Poisoned));
         assert!(!poisoned.refine_protocol_close_as_upgrade());
         assert_eq!(
             Some(CloseReason::Poisoned),
             poisoned.snapshot().close_reason
         );
         assert_eq!(Some(CloseReason::Poisoned), poisoned.probe().close_reason);
-704
+    }
-705
     #[test]
     fn committed_dispatch_drains_after_logical_close() {
         let (connection, _physical) = ConnectionState::unbounded(test_info(1));
         let dispatch = ConnectionState::try_commit_dispatch(&connection).unwrap();
         assert_eq!(ConnectionId::new(1), dispatch.connection_id());
-711
         assert!(connection.logical_close(CloseReason::ProtocolClosed));
         assert!(ConnectionState::try_commit_dispatch(&connection).is_none());
         assert_eq!(1, connection.snapshot().in_flight);
         assert_eq!(1, connection.probe().in_flight);
-715
         dispatch.complete();
         assert_eq!(0, connection.snapshot().in_flight);
         dispatch.release();
         assert_eq!(0, connection.probe().in_flight);
-718
+    }
-719
     #[test]
     fn physical_guard_is_created_once_with_the_connection() {
     fn physical_connection_guard_is_created_once_with_the_connection() {
         let (connection, physical) = ConnectionState::unbounded(test_info(1));
         assert_eq!(ConnectionId::new(1), connection.id());
         assert_eq!(PartitionId::from_index(0), connection.owner_partition());
         assert!(!connection.snapshot().physical_complete);
         physical.complete();
         assert!(connection.snapshot().physical_complete);
         assert!(!connection.probe().physical_connection_complete);
         physical.release();
         assert!(connection.probe().physical_connection_complete);
-728
+    }
-729
     #[test]
     fn connection_state_retains_immutable_connection_info() {
         #[derive(Clone, Debug, Eq, PartialEq)]
         struct ConnectorMarker(&'static str);
-734
         let origin =
             OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "example.com", None).unwrap();
         let info = ConnectionInfo::new(
             ConnectionId::new(7),
             origin.clone(),
             PartitionId::from_index(2),
             NegotiatedProtocol::Http1,
             Connected::new()
                 .proxy(true)
                 .extra(ConnectorMarker("connector-extra")),
         );
         let (connection, _physical) = ConnectionState::unbounded(info);
-747
         assert_eq!(ConnectionId::new(7), connection.info().id());
         assert_eq!(&origin, connection.info().origin());
         assert_eq!(PartitionId::from_index(2), connection.owner_partition());
         assert_eq!(NegotiatedProtocol::Http1, connection.info().protocol());
         assert_eq!(None, connection.info().local_addr());
         assert_eq!(None, connection.info().remote_addr());
         assert!(connection.info().is_proxied());
         let mut extensions = Extensions::new();
         connection.info().apply_connector_extras(&mut extensions);
         assert_eq!(
             Some(&ConnectorMarker("connector-extra")),
             extensions.get::<ConnectorMarker>()
         );
-761
+    }
-762
     #[test]
     fn root_io_drop_completes_physical_lifetime() {
     fn dropping_connection_io_completes_physical_connection_ownership() {
         let (connection, physical) = ConnectionState::unbounded(test_info(1));
         let io = ConnectionIo::new("transport", physical);
         assert_eq!(&"transport", io.get_ref());
         assert!(!connection.snapshot().physical_complete);
         assert!(!connection.probe().physical_connection_complete);
-769
         drop(io);
-771
         assert!(connection.snapshot().physical_complete);
         assert!(connection.probe().physical_connection_complete);
-773
+    }
-774
+}
-775
 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use crate::client::pool::admission::OriginAdmission;
     use std::num::NonZeroUsize;
-781
     fn test_info(id: u64) -> Arc<ConnectionInfo> {
         ConnectionInfo::for_test(ConnectionId::new(id), PartitionId::from_index(0))
-784
+    }
-785
     #[test]
     fn dispatch_commit_linearizes_against_close() {
         loom::model(|| {
             let (connection, _physical) = ConnectionState::unbounded(test_info(1));
-790
             let dispatch_connection = connection.clone();
             let dispatch = loom::thread::spawn(move || {
                 ConnectionState::try_commit_dispatch(&dispatch_connection)
             });
             let close_connection = connection.clone();
             let close = loom::thread::spawn(move || {
                 close_connection.logical_close(CloseReason::ProtocolClosed)
             });
-799
             let dispatch = dispatch.join().unwrap();
             assert!(close.join().unwrap());
             let snapshot = connection.snapshot();
             assert_eq!(Some(CloseReason::ProtocolClosed), snapshot.close_reason);
             let probe = connection.probe();
             assert_eq!(Some(CloseReason::ProtocolClosed), probe.close_reason);
             assert!(ConnectionState::try_commit_dispatch(&connection).is_none());
             match dispatch {
                 Some(dispatch) => {
                     assert_eq!(1, snapshot.in_flight);
                     assert_eq!(1, probe.in_flight);
                     drop(dispatch);
                     assert_eq!(0, connection.snapshot().in_flight);
                     assert_eq!(0, connection.probe().in_flight);
-810
+                }
                 None => assert_eq!(0, snapshot.in_flight),
                 None => assert_eq!(0, probe.in_flight),
-812
+            }
         });
-814
+    }
-815
     #[test]
     fn concurrent_logical_close_releases_one_capacity_lease() {
         loom::model(|| {
             let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
             let lease = OriginAdmission::lease_for_test(&origin);
             let (connection, _physical) = ConnectionState::bounded(test_info(1), lease);
             let first_connection = connection.clone();
             let first =
                 loom::thread::spawn(move || first_connection.logical_close(CloseReason::Poisoned));
             let second_connection = connection.clone();
             let second = loom::thread::spawn(move || {
                 second_connection.logical_close(CloseReason::PoolDropped)
             });
-829
             let first = first.join().unwrap();
             let second = second.join().unwrap();
             assert_ne!(first, second);
             assert_eq!(1, origin.available_capacity_for_test());
             let reason = connection.snapshot().close_reason.unwrap();
             let reason = connection.probe().close_reason.unwrap();
             assert!(matches!(
                 reason,
                 CloseReason::Poisoned | CloseReason::PoolDropped
             ));
         });
-840
+    }
-841
+}

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

@@ -1,1 +384,417 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Request routing and protocol dispatch.
 //!
 //! This module is the request entry point after a [`super::Client`] has
 //! resolved its partition. It rejects unsupported request forms, resolves the
 //! origin cell and partition runtime, and owns acquisition across protocol
 //! selection. Child modules prepare protocol wire forms and retain response
 //! ownership.
 //! A [`super::Client`] supplies its partition and one request. This module then
 //! performs the request journey in four phases:
 //!
 //! 1. validate protocol-sensitive request semantics;
 //! 2. resolve the origin cell and its owner runtime;
 //! 3. acquire the next authority that can make progress; and
 //! 4. submit through the selected HTTP protocol.
 //!
 //! Acquisition does not necessarily acquire a connection. A request may
 //! receive an exclusive HTTP/1 sender, an activation on an existing HTTP/2
 //! generation, or authority to establish a new transport. A waiter event is
 //! the lower-level transition that supplies one of those outcomes.
 //!
 //! Protocol dispatch may return the original request only when Hyper did not
 //! accept it. The same acquisition remains authoritative while the dispatcher
 //! selects a replacement; this is not an SDK request retry. Child modules own
 //! protocol wire preparation and accepted-response lifetime.
-25
 mod h1;
 mod h2;
-28
 use self::h1::H1DispatchResult;
 use self::h2::H2DispatchResult;
 use self::h1::H1DispatchOutcome;
 use self::h2::H2DispatchOutcome;
 use super::admission::ProtocolRequirement;
 use super::cell::{AcquisitionEvent, AcquisitionResult, OriginCell, WaiterId};
 use super::cell::h1::H1Selection;
 use super::cell::h2::H2Activation;
 use super::cell::{AcquisitionOutcome, AcquisitionStep, OriginCell, WaiterId};
 use super::establish::{self, TransportTimeout};
 use super::partition::DriverSpawner;
 use super::registry::PartitionState;
 use super::{ConnectionPool, PoolInner};
 use crate::sync::Arc;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use http_1x::{header, Method, Request, Response, Uri, Version};
 use std::future::poll_fn;
 use std::sync::Arc as StdArc;
-45
 /// Replacement selections allowed after the initial HTTP/2 dispatch attempt.
 const MAX_H2_REACQUISITIONS: usize = 2;
 const MAX_H2_REPLACEMENTS: usize = 2;
-+
 /// Operation settings known before the request's origin cell is resolved.
 ///
 /// This value is separate from [`AcquisitionContext`], which also contains
 /// cell and runtime state discovered by dispatch.
 #[derive(Clone, Debug, Default)]
 pub(super) struct RequestOptions {
     /// Transport connection timeout configured for this operation.
     connect_timeout: Option<TransportTimeout>,
-+
+}
-+
 impl RequestOptions {
     /// Creates operation settings for one pool request.
     pub(super) fn new(connect_timeout: Option<TransportTimeout>) -> Self {
         Self { connect_timeout }
-+
+    }
-+
+}
-65
 /// Stable state shared by acquisition and protocol dispatch for one request.
 ///
 /// This context owns ordinary shared configuration and identities. Linear
 /// acquisition values, such as an establishment permit or an exclusive
 /// protocol handle, remain separate arguments so their transfer is explicit.
 #[derive(Clone)]
 pub(super) struct AcquisitionContext {
     /// Shared pool policy and transport construction.
     pub(super) pool: Arc<PoolInner>,
     /// Partition from which the request was issued.
     pub(super) partition: Arc<PartitionState>,
     /// Partition-local state for the request's canonical origin.
     pub(super) cell: Arc<OriginCell>,
     /// Absolute request URI retained across protocol wire-form changes.
     pub(super) absolute_uri: Uri,
     /// Runtime that owns establishment, drivers, and pending return work.
     pub(super) owner_spawner: StdArc<dyn DriverSpawner>,
     /// Transport connection timeout configured for this operation.
     pub(super) connect_timeout: Option<TransportTimeout>,
-85
+}
-86
 impl ConnectionPool {
     /// Resolves common acquisition state and dispatches one request.
     pub(super) async fn send_request(
         &self,
 /// Resolves request-scoped pool state and runs acquisition through dispatch.
 pub(super) async fn send(
     pool: &ConnectionPool,
     partition: Arc<PartitionState>,
     request: Request<SdkBody>,
         connect_timeout: Option<TransportTimeout>,
     ) -> Result<Response<SdkBody>, ConnectorError> {
     options: RequestOptions,
 ) -> Result<Response<SdkBody>, ConnectorError> {
     validate_request_before_acquisition(&request)
         .map_err(|error| ConnectorError::user(error.into()))?;
-96
     let absolute_uri = request.uri().clone();
         let cell = self
     let cell = pool
         .inner
         .registry
         .resolve_cell(&partition, &absolute_uri)
         .map_err(|error| ConnectorError::user(error.into()))?;
     tracing::trace!(
         request_partition = ?partition.id(),
         origin_scheme = %cell.id().origin().scheme(),
         origin_host = cell.id().origin().host(),
         origin_port = ?cell.id().origin().port(),
         "request resolved to connection-pool cell"
     );
-110
     let owner_spawner = partition
         .owner_spawner()
         .map_err(|error| ConnectorError::user(error.into()))?;
     partition.ensure_maintenance_started(owner_spawner.as_ref());
-115
     let context = AcquisitionContext {
             pool: self.inner.clone(),
         pool: pool.inner.clone(),
         partition,
         cell,
         absolute_uri,
         owner_spawner,
             connect_timeout,
         connect_timeout: options.connect_timeout,
     };
-124
         self.dispatch_acquired(context, request).await
--
+    }
     acquire_and_dispatch(context, request).await
-+
+}
-+
 /// Request authority ready for one protocol-specific dispatch attempt.
 enum DispatchTarget {
     /// Exclusive HTTP/1 sender ownership.
     H1(H1Selection),
     /// Prospective request reservation on one exact HTTP/2 generation.
     H2(H2Activation),
-+
+}
-135
     /// Acquires a compatible protocol value and dispatches the request.
     ///
     /// Reacquisition is allowed only while a protocol dispatcher returns the
     /// original request envelope.
     async fn dispatch_acquired(
         &self,
 /// Acquires protocol dispatch authority and submits the request to Hyper.
 ///
 /// A protocol dispatcher may return the original request only when Hyper did
 /// not accept it. After one HTTP/2 target reaches dispatch, at most two stale
 /// replacements may return the request for another selection.
 async fn acquire_and_dispatch(
     context: AcquisitionContext,
     mut request: Request<SdkBody>,
     ) -> Result<Response<SdkBody>, ConnectorError> {
 ) -> Result<Response<SdkBody>, ConnectorError> {
     let requirement = protocol_requirement(&request);
         let mut h2_reacquisitions = H2ReacquisitionBudget::default();
     let mut h2_replacements = H2ReplacementBudget::default();
-147
     loop {
             match Self::acquire(&context, requirement).await? {
                 AcquisitionResult::H1(selection) => {
                     match self.dispatch_h1(&context, request, selection).await? {
                         H1DispatchResult::Response(response) => return Ok(response),
                         H1DispatchResult::Reacquire(returned) => request = returned,
--
+                    }
--
+                }
                 AcquisitionResult::H2(activation) => {
                     match self.dispatch_h2(&context, request, activation).await? {
                         H2DispatchResult::Response(response) => return Ok(response),
                         H2DispatchResult::Reacquire(reacquisition) => {
                             let (returned, error) = reacquisition.into_parts();
                             if !h2_reacquisitions.admit_replacement() {
         match acquire_for_dispatch(&context, requirement).await? {
             DispatchTarget::H1(selection) => {
                 match h1::dispatch(&context, request, selection).await? {
                     H1DispatchOutcome::Response(response) => return Ok(response),
                     H1DispatchOutcome::NotAccepted(returned) => request = returned,
-+
+                }
-+
+            }
             DispatchTarget::H2(activation) => {
                 match h2::dispatch(&context, request, activation).await? {
                     H2DispatchOutcome::Response(response) => return Ok(response),
                     H2DispatchOutcome::NotAccepted(unaccepted) => {
                         let (returned, error) = unaccepted.into_parts();
                         if !h2_replacements.admit_replacement() {
                             return Err(error);
-163
+                        }
                         request = returned;
-165
+                    }
-166
+                }
-167
+            }
                 AcquisitionResult::Failed(_) => {
                     unreachable!("failed acquisition returned as a successful result")
--
+                }
                 AcquisitionResult::Reacquire => {
                     unreachable!("internal reacquisition escaped the acquisition loop")
--
+                }
--
+            }
-168
+        }
-169
+    }
-+
+}
-171
     /// Acquires one protocol value compatible with the request.
     ///
     /// Local reusable values complete immediately. Otherwise one waiter
     /// remains registered while it receives a reusable protocol value, an
     /// establishment failure, or authority to start establishment.
     async fn acquire(
 /// Acquires the next protocol value that can dispatch this request.
 ///
 /// Local reusable values complete immediately under the cell lock. On a miss,
 /// one waiter remains registered while reuse and establishment race to supply
 /// dispatch authority. `poll_waiter` performs one synchronous state poll; no
 /// cell lock is retained while this future is pending.
 async fn acquire_for_dispatch(
     context: &AcquisitionContext,
     requirement: ProtocolRequirement,
     ) -> Result<AcquisitionResult, ConnectorError> {
 ) -> Result<DispatchTarget, ConnectorError> {
     'acquire: loop {
         if requirement.accepts_h2() {
             if let Some(activation) = OriginCell::select_h2(&context.cell) {
                 tracing::trace!(
                     connection_id = %activation.connection().id(),
                     request_partition = ?context.partition.id(),
                     connection_partition = ?activation.connection().owner_partition(),
                     origin_scheme = %context.cell.id().origin().scheme(),
                     origin_host = context.cell.id().origin().host(),
                     origin_port = ?context.cell.id().origin().port(),
                     "HTTP/2 pool hit; activating local generation"
                 );
                     return Ok(AcquisitionResult::H2(activation));
                 return Ok(DispatchTarget::H2(activation));
-195
+            }
-196
+        }
         if requirement.accepts_h1() {
             if let Some(selection) = OriginCell::select_h1(&context.cell) {
                 tracing::trace!(
                     connection_id = %selection.connection_id(),
                     request_partition = ?context.partition.id(),
                     connection_partition = ?selection.connection().owner_partition(),
                     origin_scheme = %selection.connection().info().origin().scheme(),
                     origin_host = selection.connection().info().origin().host(),
                     origin_port = ?selection.connection().info().origin().port(),
                     "HTTP/1 pool hit; reusing idle connection"
                 );
                     return Ok(AcquisitionResult::H1(selection));
                 return Ok(DispatchTarget::H1(selection));
-209
+            }
-210
+        }
-211
         let waiter = OriginCell::register_waiter(&context.cell, requirement);
         tracing::trace!(
             request_partition = ?context.partition.id(),
             origin_scheme = %context.cell.id().origin().scheme(),
             origin_host = context.cell.id().origin().host(),
             origin_port = ?context.cell.id().origin().port(),
             protocol_requirement = ?requirement,
             "connection acquisition queued"
         );
             let mut waiter_guard = WaiterGuard::new(context.cell.clone(), waiter);
         let mut waiter_guard = WaiterCancellationGuard::new(context.cell.clone(), waiter);
         loop {
             match poll_fn(|cx| context.cell.poll_waiter(waiter, cx)).await {
                     AcquisitionEvent::Complete(AcquisitionResult::H1(selection)) => {
                 AcquisitionStep::Resolved(AcquisitionOutcome::H1(selection)) => {
                     waiter_guard.disarm();
                         return Ok(AcquisitionResult::H1(selection));
                     return Ok(DispatchTarget::H1(selection));
-227
+                }
                     AcquisitionEvent::Complete(AcquisitionResult::H2(activation)) => {
                 AcquisitionStep::Resolved(AcquisitionOutcome::H2(activation)) => {
                     waiter_guard.disarm();
                         OriginCell::service_h2_waiters(&context.cell);
                         OriginCell::service_peer_h2_waiters(&context.cell);
                     OriginCell::offer_local_h2(&context.cell);
                     OriginCell::offer_peer_h2(&context.cell);
                     tracing::trace!(
                         connection_id = %activation.connection().id(),
                         request_partition = ?context.partition.id(),
                         connection_partition = ?activation.connection().owner_partition(),
                         origin_scheme = %context.cell.id().origin().scheme(),
                         origin_host = context.cell.id().origin().host(),
                         origin_port = ?context.cell.id().origin().port(),
                         "HTTP/2 acquisition completed"
                     );
                         return Ok(AcquisitionResult::H2(activation));
                     return Ok(DispatchTarget::H2(activation));
-242
+                }
                     AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) => {
                 AcquisitionStep::Resolved(AcquisitionOutcome::Failed(error)) => {
                     waiter_guard.disarm();
                     return Err(error);
-246
+                }
                     AcquisitionEvent::Complete(AcquisitionResult::Reacquire) => {
                 AcquisitionStep::Resolved(AcquisitionOutcome::RetryAcquisition) => {
                     waiter_guard.disarm();
                     continue 'acquire;
-250
+                }
                     AcquisitionEvent::Establish(permit) => {
                 AcquisitionStep::StartEstablishment(permit) => {
                     tracing::trace!(
                         request_partition = ?context.partition.id(),
                         origin_scheme = %context.cell.id().origin().scheme(),
                         origin_host = context.cell.id().origin().host(),
                         origin_port = ?context.cell.id().origin().port(),
                         "connection establishment starting"
                     );
                     let attempt =
                         establish::establish(context.clone(), waiter, permit, requirement);
                     let completion =
                         EstablishmentCompletionGuard::new(context.cell.clone(), waiter);
                     context.owner_spawner.spawn(Box::pin(async move {
                         let mut completion = completion;
                         if !completion.start() {
                             drop(attempt);
                             completion.disarm();
                             return;
-269
+                        }
                         match attempt.await {
                             establish::EstablishmentOutcome::Complete(result) => {
                                 completion.complete(result);
-273
+                            }
                                 establish::EstablishmentOutcome::Transferred => completion.disarm(),
                             establish::EstablishmentOutcome::WaiterCompletionTransferred => {
                                 completion.disarm();
-+
+                            }
-277
+                        }
                     }));
-279
+                }
-280
+            }
-281
+        }
-282
+    }
--
+    }
-283
+}
--
 /// Per-request bound on replacement HTTP/2 selections.
 #[derive(Default)]
 struct H2ReacquisitionBudget {
 struct H2ReplacementBudget {
     /// Replacement selections admitted after the initial selection.
     completed: usize,
     used: usize,
-289
+}
-290
 impl H2ReacquisitionBudget {
 impl H2ReplacementBudget {
     /// Returns whether one more replacement selection may proceed.
     fn admit_replacement(&mut self) -> bool {
         if self.completed >= MAX_H2_REACQUISITIONS {
         if self.used >= MAX_H2_REPLACEMENTS {
             return false;
-296
+        }
         self.completed += 1;
         self.used += 1;
         true
-299
+    }
-300
+}
-301
 /// Guarantees a terminal result for one submitted establishment attempt.
 ///
 /// The guard enters the owner-runtime future before submission. Runtime task
 /// drop reports failure to the waiter. A result that wins before the submitted
 /// future starts disarms the guard without polling the connector.
 struct EstablishmentCompletionGuard {
     cell: Arc<OriginCell>,
     waiter: WaiterId,
     active: bool,
-311
+}
-312
 impl EstablishmentCompletionGuard {
     fn new(cell: Arc<OriginCell>, waiter: WaiterId) -> Self {
         Self {
             cell,
             waiter,
             active: true,
-319
+        }
-320
+    }
-321
     fn start(&self) -> bool {
         self.cell.start_establishment(self.waiter)
-324
+    }
-325
     fn complete(mut self, result: AcquisitionResult) {
     fn complete(mut self, result: AcquisitionOutcome) {
         self.active = false;
         self.cell.complete_establishment(self.waiter, result);
-329
+    }
-330
     fn disarm(&mut self) {
         self.active = false;
-333
+    }
-334
+}
-335
 impl Drop for EstablishmentCompletionGuard {
     fn drop(&mut self) {
         if self.active {
             let error = ConnectorError::io(EstablishmentTaskDropped.into());
             self.cell
                 .complete_establishment(self.waiter, AcquisitionResult::Failed(error));
                 .complete_establishment(self.waiter, AcquisitionOutcome::Failed(error));
             tracing::debug!(
                 request_partition = ?self.cell.id().partition(),
                 connection_partition = ?self.cell.id().partition(),
                 origin_scheme = %self.cell.id().origin().scheme(),
                 origin_host = self.cell.id().origin().host(),
                 origin_port = ?self.cell.id().origin().port(),
                 "connection establishment task dropped"
             );
-350
+        }
-351
+    }
-352
+}
-353
 /// The owner runtime discarded an establishment future before completion.
 #[derive(Debug)]
 struct EstablishmentTaskDropped;
-357
 impl std::fmt::Display for EstablishmentTaskDropped {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("the owner-runtime connection establishment task was dropped")
-361
+    }
-362
+}
-363
 impl std::error::Error for EstablishmentTaskDropped {}
-365
 /// Cancels a request's waiter until it consumes a terminal acquisition result.
 struct WaiterGuard {
 /// Cancels a request's waiter until it consumes a terminal acquisition outcome.
 struct WaiterCancellationGuard {
     cell: Arc<OriginCell>,
     waiter: WaiterId,
     active: bool,
-371
+}
-372
 impl WaiterGuard {
 impl WaiterCancellationGuard {
     fn new(cell: Arc<OriginCell>, waiter: WaiterId) -> Self {
         Self {
             cell,
             waiter,
             active: true,
-379
+        }
-380
+    }
-381
     fn disarm(&mut self) {
         self.active = false;
-384
+    }
-385
+}
-386
 impl Drop for WaiterGuard {
 impl Drop for WaiterCancellationGuard {
     fn drop(&mut self) {
         if self.active {
             OriginCell::cancel_waiter(&self.cell, self.waiter);
-391
+        }
-392
+    }
-393
+}
-394
 /// Rejects request forms the pool cannot route before acquiring connection state.
 fn validate_request_before_acquisition(
     request: &Request<SdkBody>,

↧ Expand context

 ) -> Result<(), RequestDispatchError> {
     if request.version() == Version::HTTP_2 && request.method() == Method::CONNECT {
         return Err(RequestDispatchError::ExtendedConnectUnsupported);
-401
+    }
     if request.version() == Version::HTTP_2 && has_upgrade_semantics(request) {
         return Err(RequestDispatchError::UpgradeOverHttp2);
-404
+    }
     match request.version() {
         Version::HTTP_11 | Version::HTTP_2 => Ok(()),
         Version::HTTP_10 if request.method() == Method::CONNECT => {
             Err(RequestDispatchError::ConnectOverHttp10)
-409
+        }
         Version::HTTP_10 => Ok(()),
         version => Err(RequestDispatchError::UnsupportedVersion(version)),
-412
+    }
-413
+}
-414
 /// Returns the protocol capability needed to preserve the request's wire semantics.
 fn protocol_requirement(request: &Request<SdkBody>) -> ProtocolRequirement {
     if request.version() == Version::HTTP_2 {

@@ -493,526 +530,563 @@

↥ Expand context

         );
-527
         let mut upgrade = request(Method::GET, Version::HTTP_11);
         upgrade
             .headers_mut()
             .insert(header::CONNECTION, "keep-alive, Upgrade".parse().unwrap());
         assert_eq!(
             ProtocolRequirement::H1Required,
             protocol_requirement(&upgrade)
         );
-536
+    }
-537
     #[test]
     fn unsupported_h2_tunnel_and_upgrade_forms_are_rejected() {
         assert!(matches!(
             validate_request_before_acquisition(&request(Method::CONNECT, Version::HTTP_2)),
             Err(RequestDispatchError::ExtendedConnectUnsupported)
         ));
-544
         let mut upgrade = request(Method::GET, Version::HTTP_2);

         upgrade
             .headers_mut()
             .insert(header::UPGRADE, "websocket".parse().unwrap());
         assert!(matches!(
             validate_request_before_acquisition(&upgrade),
             Err(RequestDispatchError::UpgradeOverHttp2)
         ));
-553
+    }
-554
     #[test]
     fn h2_reacquisition_is_bounded_after_two_replacements() {
         let mut budget = H2ReacquisitionBudget::default();
         for _ in 0..MAX_H2_REACQUISITIONS {
     fn h2_replacement_is_bounded_after_two_selections() {
         let mut budget = H2ReplacementBudget::default();
         for _ in 0..MAX_H2_REPLACEMENTS {
             assert!(budget.admit_replacement());
-560
+        }
         assert!(!budget.admit_replacement());
-562
+    }
-563
+}

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

@@ -1,1 +248,239 @@

↥ Expand context

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! HTTP/1 wire preparation, exclusive dispatch, and response completion.
 //!

 //! The parent dispatcher has already validated the request, resolved its
 //! origin cell, acquired an HTTP/1 selection, and selected the partition
 //! runtime. This module prepares the HTTP/1 wire form and owns the exclusive
 //! sender until the response reaches a reusable message boundary. Every
 //! cancellation path returns the sender to its connection cell or retires the
 //! connection.
-14
 use super::super::cell::h1::{H1Exchange, H1Selection};
 use super::super::connection::{CloseReason, ConnectionState, DispatchGuard};
 use super::super::partition::DriverSpawner;
 use super::super::ConnectionPool;
 use super::{AcquisitionContext, H1HostHeaderInserted, RequestDispatchError};
 use crate::client::downcast_error;
 use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use http_1x::{Method, Request, Response, Uri, Version};
 use hyper::body::Body;
 use std::future::poll_fn;
 use std::pin::Pin;
 use std::sync::Arc as StdArc;
 use std::task::{Context, Poll};
-29
 /// Result of one HTTP/1 dispatch attempt while the pool owns the request.
 pub(super) enum H1DispatchResult {
 pub(super) enum H1DispatchOutcome {
     /// Hyper accepted the request and produced a guarded response.
     Response(Response<SdkBody>),
     /// The selection became stale or Hyper returned the request unsent.
     Reacquire(Request<SdkBody>),
     NotAccepted(Request<SdkBody>),
-36
+}
-37
 impl ConnectionPool {
     /// Attempts one dispatch through an acquired HTTP/1 selection.
     ///
     /// The parent may reacquire only when this function returns the original
     /// request. A fresh connection's send error is terminal even if Hyper
     /// returns the request.
     pub(super) async fn dispatch_h1(
         &self,
 /// Attempts one dispatch through an acquired HTTP/1 selection.
 ///
 /// The parent may reacquire only when this function returns the original
 /// request. A fresh connection's send error is terminal even if Hyper returns
 /// the request.
 pub(super) async fn dispatch(
     context: &AcquisitionContext,
     mut request: Request<SdkBody>,
     mut selection: H1Selection,
     ) -> Result<H1DispatchResult, ConnectorError> {
 ) -> Result<H1DispatchOutcome, ConnectorError> {
     let request_method = request.method().clone();
     let reused = selection.is_reused();
     let connection = selection.connection().clone();
     let close_handle = selection.close_handle();
     let captured_metadata = request
         .extensions()
         .get::<CaptureSmithyConnection>()
         .cloned()
         .map(|capture| {
             let metadata = connection.info().metadata(close_handle.clone());
             let captured = metadata.clone();
             capture.set_connection_retriever(move || Some(captured.clone()));
             metadata
         });
-62
     if request.version() == Version::HTTP_2 {
         let metadata =
             captured_metadata.unwrap_or_else(|| connection.info().metadata(close_handle));
         return Err(ConnectorError::user(
             "an HTTP/2 request cannot use an HTTP/1 connection".into(),
-68
+        )
         .with_connection(metadata));
-70
+    }
-71
     add_host_header(&mut request, &context.absolute_uri)
         .map_err(|error| ConnectorError::user(error.into()))?;
     rewrite_h1_request_target(&mut request, connection.info().is_proxied());
-75
     // Commit against logical close before transferring the request to
     // Hyper. A close that wins this race leaves the request untouched.
     let Some(dispatch) = ConnectionState::try_commit_dispatch(&connection) else {
         tracing::trace!(
             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(),
             "HTTP/1 selection became stale before dispatch"
         );
         *request.uri_mut() = context.absolute_uri.clone();
         selection.retire_connection(CloseReason::ProtocolClosed);
             return Ok(H1DispatchResult::Reacquire(request));
         return Ok(H1DispatchOutcome::NotAccepted(request));
     };
     let send = selection.sender_mut().hyper_mut().try_send_request(request);
-93
     let exchange = selection.into_exchange();
     match send.await {
         Ok(mut response) => {
             // Response-body ownership keeps both the accepted dispatch
             // and exclusive request handle out of the pool until Hyper
             // proves a complete message boundary.
             connection
                 .info()
                 .apply_connector_extras(response.extensions_mut());
                 Ok(H1DispatchResult::Response(guard_response(
             Ok(H1DispatchOutcome::Response(guard_h1_response(
                 response,
                 request_method,
                 exchange,
                 dispatch,
                 context.owner_spawner.clone(),
             )))
-110
+        }
         Err(mut error) => {
             if let Some(mut returned) = error.take_message() {
                 exchange.retire_connection(CloseReason::ProtocolClosed);
                 drop(dispatch);
                 if reused {
                     tracing::trace!(
                         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(),
                         "reused HTTP/1 connection returned request unsent"
                     );
                     *returned.uri_mut() = context.absolute_uri.clone();
                         return Ok(H1DispatchResult::Reacquire(returned));
                     return Ok(H1DispatchOutcome::NotAccepted(returned));
-127
+                }
                     let metadata = captured_metadata
                         .unwrap_or_else(|| connection.info().metadata(close_handle));
                     return Err(
                         super::super::super::downcast_error(Box::new(error.into_error()))
                             .with_connection(metadata),
                     );
                 let metadata =
                     captured_metadata.unwrap_or_else(|| connection.info().metadata(close_handle));
                 return Err(downcast_error(Box::new(error.into_error())).with_connection(metadata));
-131
+            }
             exchange.retire_connection(CloseReason::IncompleteH1Exchange);
             drop(dispatch);
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().metadata(close_handle));
                 Err(
                     super::super::super::downcast_error(Box::new(error.into_error()))
                         .with_connection(metadata),
--
+                )
--
+            }
             Err(downcast_error(Box::new(error.into_error())).with_connection(metadata))
-137
+        }
-138
+    }
-139
+}
-140
 /// Attaches connection return and dispatch completion to an accepted response.
 ///
 /// An ordinary response retains both guards until Hyper proves a complete
 /// message boundary. Successful CONNECT and `101` responses retire the pool
 /// record before the response exposes Hyper's upgraded root I/O.
 fn guard_response(
 fn guard_h1_response(
     response: Response<hyper::body::Incoming>,
     method: Method,
     exchange: H1Exchange,
     dispatch: DispatchGuard,
     spawner: StdArc<dyn DriverSpawner>,
 ) -> Response<SdkBody> {
     let upgrade = response.status() == http_1x::StatusCode::SWITCHING_PROTOCOLS
         || (method == Method::CONNECT && response.status().is_success());
     let (parts, body) = response.into_parts();
     if upgrade {
         exchange.retire_connection(CloseReason::Upgraded);
         dispatch.complete();
         dispatch.release();
         return Response::from_parts(parts, SdkBody::from_body_1_x(body));
-160
+    }
     let body = H1ResponseBody::new(body, exchange, dispatch, spawner);
     Response::from_parts(parts, SdkBody::from_body_1_x(body))
-163
+}
-164
 /// Response body that retains the HTTP/1 exchange until its message boundary.
 ///
 /// End-of-stream returns a ready request handle to pool policy. A body error,
 /// or dropping the body before Hyper reports end-of-stream, retires the
 /// connection instead.
 struct H1ResponseBody {
     /// Hyper response body whose terminal state determines message completion.
     inner: hyper::body::Incoming,
     /// Sender and dispatch ownership until one terminal body observation.
     lifecycle: Option<H1ResponseLifecycle>,
-175
+}
-176
 impl H1ResponseBody {
     /// Wraps a response and immediately completes a body already at end stream.
     fn new(
         inner: hyper::body::Incoming,
         exchange: H1Exchange,
         dispatch: DispatchGuard,
         spawner: StdArc<dyn DriverSpawner>,
     ) -> Self {
         let mut body = Self {
             inner,
             lifecycle: Some(H1ResponseLifecycle {
                 exchange: Some(exchange),
                 dispatch: Some(dispatch),
                 spawner,
             }),
         };
         if body.inner.is_end_stream() {
             body.finish_without_context();
-195
+        }
         body
-197
+    }
-198
     /// Completes response ownership using the current body-task waker.
     fn finish_with_context(&mut self, cx: &mut Context<'_>) {
         if let Some(lifecycle) = self.lifecycle.take() {
             lifecycle.finish(Some(cx));
             lifecycle.resolve(Some(cx));
-203
+        }
-204
+    }
-205
     /// Completes response ownership without a task context when already ready.
     fn finish_without_context(&mut self) {
         if let Some(lifecycle) = self.lifecycle.take() {
             lifecycle.finish(None);
             lifecycle.resolve(None);
-210
+        }
-211
+    }
-212
     /// Retires an exchange whose response body ended with an error.
     fn fail(&mut self) {
         if let Some(mut lifecycle) = self.lifecycle.take() {
             if let Some(exchange) = lifecycle.exchange.take() {
                 exchange.retire_connection(CloseReason::IncompleteH1Exchange);
-218
+            }
             drop(lifecycle.dispatch.take());

↧ Expand context

-220
+        }
-221
+    }
-222
+}
-223
 impl Body for H1ResponseBody {
     type Data = <hyper::body::Incoming as Body>::Data;
     type Error = <hyper::body::Incoming as Body>::Error;
-227
     fn poll_frame(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
         match Pin::new(&mut self.inner).poll_frame(cx) {
             Poll::Ready(None) => {
                 self.finish_with_context(cx);
                 Poll::Ready(None)
-236
+            }
             Poll::Ready(Some(Err(error))) => {
                 self.fail();
                 Poll::Ready(Some(Err(error)))

@@ -278,269 +366,357 @@

↥ Expand context

 /// [`H1Exchange`] and [`DispatchGuard`], retiring the incomplete exchange and
 /// completing dispatch accounting through their fallbacks.
 struct H1ResponseLifecycle {
     /// Exclusive request handle retained until Hyper proves it reusable.
     exchange: Option<H1Exchange>,
     /// Accepted-dispatch accounting completed with response cleanup.
     dispatch: Option<DispatchGuard>,
     /// Owner-runtime placement for readiness work that outlives the body.
     spawner: StdArc<dyn DriverSpawner>,
-278
+}
-279
 /// Owner-runtime task waiting for Hyper to prove the sender reusable.
 ///
 /// [`H1ResponseLifecycle`] creates this only after the response reaches
 /// end-of-stream while sender readiness is still pending. Readiness success
 /// returns the sender through [`H1Exchange::offer_for_reuse`]; readiness failure
 /// retires the connection. Dropping the submitted task before either result retires the
 /// exchange as [`CloseReason::OwnerRuntimeShutdown`], so an unproven sender can
 /// never re-enter the pool.
 struct H1ReadinessTask {

     /// Exchange retired as owner-runtime shutdown if the task is dropped.
     exchange: Option<H1Exchange>,
-291
+}
-292
 impl H1ResponseLifecycle {
     /// Returns or retires the request handle and completes dispatch accounting.
     ///
     /// Readiness is first polled with the response body's task context when one
     /// is available. Pending readiness moves to an owner-runtime task so the
     /// completed body does not retain connection ownership.
     fn finish(mut self, cx: Option<&mut Context<'_>>) {
     fn resolve(mut self, cx: Option<&mut Context<'_>>) {
         let mut exchange = self
             .exchange
             .take()
             .expect("HTTP/1 response lifecycle lost its exchange");
         let ready = match cx {
             Some(cx) => exchange.poll_ready(cx),
             None if exchange.is_ready() => Poll::Ready(Ok(())),
             None => Poll::Pending,
         };
         match ready {
             Poll::Ready(Ok(())) => {
                 exchange.offer_for_reuse();
-312
+            }
             Poll::Ready(Err(_)) => {
                 exchange.retire_connection(CloseReason::ProtocolClosed);
-315
+            }
             Poll::Pending => {
                 let mut readiness = H1ReadinessTask::new(exchange);
                 self.spawner.spawn(Box::pin(async move {
                     match poll_fn(|cx| readiness.poll_ready(cx)).await {
                         Ok(()) => readiness.reuse(),
                         Err(_) => readiness.retire_connection(CloseReason::ProtocolClosed),
-322
+                    }
                 }));
-324
+            }
-325
+        }
         if let Some(dispatch) = self.dispatch.take() {
             dispatch.complete();
             dispatch.release();
-328
+        }
-329
+    }
-330
+}
-331
 impl H1ReadinessTask {
     /// Arms owner-runtime fallback around one pending exchange.
     fn new(exchange: H1Exchange) -> Self {
         Self {
             exchange: Some(exchange),
-337
+        }

↧ Expand context

-338
+    }
-339
     /// Polls Hyper for proof that another request may be sent.
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), hyper::Error>> {
         self.exchange
             .as_mut()
             .expect("HTTP/1 readiness task consumed more than once")
             .poll_ready(cx)
-346
+    }
-347
     /// Returns the ready handle through ordinary owning-cell arbitration.
     fn reuse(mut self) {
         let exchange = self
             .exchange
             .take()
             .expect("HTTP/1 readiness task consumed more than once");
         exchange.offer_for_reuse();
-355
+    }
-356
     /// Retires the connection after a terminal readiness result.

@@ -408,399 +467,459 @@

↥ Expand context

     request.extensions_mut().insert(H1HostHeaderInserted);
     Ok(())
-401
+}
-402
 /// Rewrites an absolute URI into the request target required by HTTP/1.
 fn rewrite_h1_request_target(request: &mut Request<SdkBody>, is_proxied: bool) {
     use http_1x::uri::Parts;
-406
     if request.method() == Method::CONNECT {
         if let Some(authority) = request.uri().authority().cloned() {
             let mut parts = Parts::default();
             parts.authority = Some(authority);
             *request.uri_mut() = Uri::from_parts(parts).expect("request authority is a valid URI");
-412
+        }
         return;
-414
+    }
     if is_proxied {
         return;
-417
+    }
     let mut parts = Parts::default();

     parts.path_and_query = request.uri().path_and_query().cloned();
     *request.uri_mut() = Uri::from_parts(parts).expect("request path and query are a valid URI");
-421
+}
-422
 #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
 mod tests {
     use super::*;
     use crate::client::pool::cell::h1::H1Sender;
     use crate::client::pool::cell::OriginCell;
     use crate::client::pool::connection::ConnectionInfo;
     use crate::client::pool::dispatch::RequestOptions;
     use crate::client::pool::registry::PartitionState;
     use crate::client::pool::{
         Client, ConnectionId, ConnectionPool, ConnectionReuseScope, Partition, PartitionId,
         TokioDriverSpawner,
     };
     use crate::client::timeout::test::{NeverConnects, NeverReplies};
     use crate::sync::Arc;
     use aws_smithy_async::rt::sleep::TokioSleep;
     use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
     use aws_smithy_runtime_api::client::http::{HttpClient, HttpConnector, HttpConnectorSettings};

↧ Expand context

     use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
     use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
     use http_body_util::BodyExt;
     use hyper_util::client::legacy::connect::{Connected, Connection};
     use hyper_util::rt::TokioIo;
     use std::future::Future;
     use std::io::{self, IoSlice};
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::task::{Context, Poll};
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
     use tokio::net::{TcpListener, TcpStream};
     use tokio::sync::Notify;
     use tower::Service;
-453
     #[derive(Debug)]
     struct DroppingSpawner;
-456
     impl DriverSpawner for DroppingSpawner {
         fn spawn(&self, driver: Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>) {
             drop(driver);

@@ -605,597 +692,684 @@

↥ Expand context

-597
         fn call(&mut self, uri: Uri) -> Self::Future {
             let address = uri
                 .authority()
                 .expect("test URI has an authority")
                 .as_str()
                 .to_owned();
             Box::pin(async move {
                 TcpStream::connect(address)
                     .await
                     .map(TokioIo::new)
                     .map(ExtraIo)
             })
-610
+        }
-611
+    }
-612
     #[test]
     fn dropped_readiness_task_records_owner_runtime_shutdown() {
         let pool = ConnectionPool::builder()
             .idle_timeout(None)

             .build_http()
             .unwrap();
         let partition = anonymous_partition(&pool);
         let uri = "http://example.com/".parse().unwrap();
         let cell = pool.inner.registry.resolve_cell(&partition, &uri).unwrap();
         let (connection, _physical) = ConnectionState::unbounded(ConnectionInfo::for_test(
             ConnectionId::new(1),
             PartitionId::ANONYMOUS,
         ));
         let selection =
             OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
             OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
-628
         drop(H1ReadinessTask::new(selection.into_exchange()));
-630
         assert_eq!(
             Some(CloseReason::OwnerRuntimeShutdown),
             connection.snapshot().close_reason
             connection.probe().close_reason
         );
-635
+    }
-636
     #[tokio::test]
     async fn dropped_owner_task_completes_the_establishment_waiter() {
         let pool = ConnectionPool::builder()
             .idle_timeout(None)
             .partitions([Partition::new(PartitionId::from_index(7), DroppingSpawner)])
             .build_http()
             .unwrap();
         let partition = pool
             .inner
             .registry
             .partition(PartitionId::from_index(7))
             .unwrap();
         let request = Request::get("http://127.0.0.1:9/")
             .body(SdkBody::empty())
             .unwrap();
-652
         let error = pool
             .send_request(partition, request, None)
             .send_request(partition, request, RequestOptions::default())
             .await
             .expect_err("dropped establishment task did not fail the request");
         assert!(error.is_io());
         assert_eq!(
             "the owner-runtime connection establishment task was dropped",
             std::error::Error::source(&error).unwrap().to_string()
         );
-662
+    }
-663
     struct TestServer {

↧ Expand context

         endpoint: String,
         accepted: StdArc<AtomicUsize>,
         task: tokio::task::JoinHandle<()>,
-668
+    }
-669
     impl TestServer {
         async fn start() -> Self {
             let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
             let endpoint = format!("http://{}", listener.local_addr().unwrap());
             let accepted = StdArc::new(AtomicUsize::new(0));
             let accepted_for_task = accepted.clone();
             let task = tokio::spawn(async move {
                 while let Ok((stream, _)) = listener.accept().await {
                     accepted_for_task.fetch_add(1, Ordering::SeqCst);
                     tokio::spawn(serve_connection(stream));
-680
+                }
             });
             Self {
                 endpoint,
                 accepted,

@@ -734,726 +794,786 @@

↥ Expand context

     async fn read_request_head(stream: &mut TcpStream) -> bool {
         let mut buffered = Vec::new();
         let mut read = [0_u8; 1024];
         loop {
             let count = match stream.read(&mut read).await {
                 Ok(0) | Err(_) => return false,
                 Ok(count) => count,
             };
             buffered.extend_from_slice(&read[..count]);
             if buffered.windows(4).any(|window| window == b"\r\n\r\n") {
                 return true;
-737
+            }
-738
+        }
-739
+    }
-740
     fn anonymous_partition(pool: &ConnectionPool) -> Arc<PartitionState> {
         pool.inner
             .registry
             .partition(PartitionId::ANONYMOUS)
             .unwrap()

-746
+    }
-747
     async fn request(
         pool: &ConnectionPool,
         partition: Arc<PartitionState>,
         uri: Uri,
     ) -> Response<SdkBody> {
         pool.send_request(
             partition,
             Request::get(uri).body(SdkBody::empty()).unwrap(),
             None,
             RequestOptions::default(),
-757
+        )
         .await
         .unwrap()
-760
+    }
-761
     async fn consume(response: Response<SdkBody>) -> hyper::body::Bytes {
         response.into_body().collect().await.unwrap().to_bytes()
-764
+    }
-765
     #[test]

↧ Expand context

     fn host_header_preserves_or_derives_the_authority() {
         let cases = [
-769
+            (
                 "existing",
                 "http://ignored.example/",
                 Some("kept.example"),
                 "kept.example",
             ),
-775
+            (
                 "implicit default port",
                 "http://example.com/",
                 None,
                 "example.com",
             ),
-781
+            (
                 "HTTP default port",
                 "http://example.com:80/",
                 None,
                 "example.com",
             ),

@@ -918,910 +1007,999 @@

↥ Expand context

                 PartitionId::from_index(7),
                 CountingSpawner {
                     submitted: submitted.clone(),
                 },
             )])
             .build_http()
             .unwrap();
         let partition = pool
             .inner
             .registry
             .partition(PartitionId::from_index(7))
             .unwrap();
         let uri = server.uri("/ready");
         consume(request(&pool, partition.clone(), uri.clone()).await).await;
         let baseline = submitted.load(Ordering::SeqCst);
         let cell = pool.inner.registry.resolve_cell(&partition, &uri).unwrap();
         let selection = OriginCell::select_h1(&cell).expect("ready sender was not reusable");
         let connection = selection.connection().clone();
         let dispatch =
             ConnectionState::try_commit_dispatch(&connection).expect("connection was not open");

         let exchange = selection.into_exchange();
         assert!(exchange.is_ready());
-932
         H1ResponseLifecycle {
             exchange: Some(exchange),
             dispatch: Some(dispatch),
             spawner: StdArc::new(CountingSpawner {
                 submitted: submitted.clone(),
             }),
-939
+        }
         .finish(None);
         .resolve(None);
-941
         assert_eq!(baseline, submitted.load(Ordering::SeqCst));
-943
+    }
-944
     #[tokio::test]
     async fn cancelling_launched_establishment_returns_its_connection_to_a_later_waiter() {
         let server = TestServer::start().await;
         let started = StdArc::new(Notify::new());
         let release = StdArc::new(Notify::new());
         let calls = StdArc::new(AtomicUsize::new(0));
         let pool = super::super::super::builder::Builder::default()
             .idle_timeout(None)
             .max_connections_per_host(1)
             .build_with_connector(PausedConnector {
                 started: started.clone(),
                 release: release.clone(),
                 calls: calls.clone(),
             })
             .unwrap();
         let partition = anonymous_partition(&pool);
         let first_pool = pool.clone();
         let first_partition = partition.clone();
         let first_uri = server.uri("/cancelled-establishment");
         let first = tokio::spawn(async move {
             first_pool
                 .send_request(
                     first_partition,
                     Request::get(first_uri).body(SdkBody::empty()).unwrap(),
                     None,
                     RequestOptions::default(),
-970
+                )
                 .await
         });
         started.notified().await;
-974
         first.abort();
         assert!(first.await.unwrap_err().is_cancelled());
         release.notify_one();
-978
         let response = tokio::time::timeout(

↧ Expand context

             std::time::Duration::from_secs(1),
             request(
                 &pool,
                 partition,
                 server.uri("/after-cancelled-establishment"),
             ),
-986
+        )
         .await
         .expect("completed establishment did not satisfy the later waiter");
         consume(response).await;
         assert_eq!(1, calls.load(Ordering::SeqCst));
         assert_eq!(1, server.accepted());
-992
+    }
-993
     #[tokio::test]
     async fn cancelling_an_accepted_exchange_retires_its_connection() {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
         let endpoint = format!("http://{}", listener.local_addr().unwrap());
         let accepted = StdArc::new(AtomicUsize::new(0));
         let accepted_for_task = accepted.clone();

@@ -1010,1002 +1086,1078 @@

↥ Expand context

         let server = tokio::spawn(async move {
             while let Ok((mut stream, _)) = listener.accept().await {
                 let index = accepted_for_task.fetch_add(1, Ordering::SeqCst);
                 if index == 0 {
                     let first_request = first_request_for_task.clone();
                     tokio::spawn(async move {
                         assert!(read_request_head(&mut stream).await);
                         first_request.notify_one();
                         let mut sink = [0_u8; 32];
                         while stream.read(&mut sink).await.unwrap_or(0) != 0 {}
                     });
                 } else {
                     tokio::spawn(serve_connection(stream));
-1015
+                }
-1016
+            }
         });
         let pool = ConnectionPool::builder()
             .idle_timeout(None)
             .max_connections_per_host(1)
             .build_http()

             .unwrap();
         let partition = anonymous_partition(&pool);
         let first_pool = pool.clone();
         let first_partition = partition.clone();
         let first_uri: Uri = format!("{endpoint}/cancelled-exchange").parse().unwrap();
         let first = tokio::spawn(async move {
             first_pool
                 .send_request(
                     first_partition,
                     Request::get(first_uri).body(SdkBody::empty()).unwrap(),
                     None,
                     RequestOptions::default(),
-1033
+                )
                 .await
         });
         first_request.notified().await;
         let cell = pool
             .inner
             .registry
             .resolve_cell(&partition, &format!("{endpoint}/").parse().unwrap())
             .unwrap();
         let connection = cell.only_h1_connection_for_test();
-1043
         first.abort();
         assert!(first.await.unwrap_err().is_cancelled());
         assert_eq!(
             Some(CloseReason::IncompleteH1Exchange),
             connection.snapshot().close_reason
             connection.probe().close_reason
         );
-1050
         let second_uri: Uri = format!("{endpoint}/after-cancelled-exchange")
             .parse()
             .unwrap();
         let second = tokio::time::timeout(
             std::time::Duration::from_secs(1),
             request(&pool, partition, second_uri),
-1057
+        )
         .await

↧ Expand context

         .expect("accepted-exchange cancellation did not release bounded capacity");
         consume(second).await;
         assert_eq!(2, accepted.load(Ordering::SeqCst));
         server.abort();
-1063
+    }
-1064
     #[tokio::test]
     async fn bounded_second_request_waits_for_h1_return() {
         let server = TestServer::start().await;
         let pool = ConnectionPool::builder()
             .max_connections_per_host(1)
             .build_http()
             .unwrap();
         let partition = anonymous_partition(&pool);
         let first = request(&pool, partition.clone(), server.uri("/first")).await;
-1074
         let second_pool = pool.clone();
         let second_partition = partition.clone();
         let second_uri = server.uri("/second");
         let second =

@@ -1143,1135 +1311,1303 @@

↥ Expand context

     #[tokio::test]
     async fn dropping_incomplete_body_retires_connection() {
         let server = TestServer::start().await;
         let pool = ConnectionPool::builder().build_http().unwrap();
         let partition = anonymous_partition(&pool);
-1140
         drop(request(&pool, partition.clone(), server.uri("/first")).await);
         assert_eq!(
             hyper::body::Bytes::from_static(b"ok"),
             consume(request(&pool, partition, server.uri("/second")).await).await
         );
         assert_eq!(2, server.accepted());
-1147
+    }
-1148
     #[tokio::test]
     async fn incompatible_h1_result_remains_reusable_for_compatible_demand() {
         let server = TestServer::start().await;
         let pool = ConnectionPool::builder()
             .max_connections_per_host(1)
             .build_http()

             .unwrap();
         let partition = anonymous_partition(&pool);
         let capture = CaptureSmithyConnection::new();
         let mut h2_request = Request::get(server.uri("/h2"))
             .version(Version::HTTP_2)
             .body(SdkBody::empty())
             .unwrap();
         h2_request.extensions_mut().insert(capture.clone());
-1163
         let error = pool
             .send_request(partition.clone(), h2_request, None)
             .send_request(partition.clone(), h2_request, RequestOptions::default())
             .await
             .expect_err("an HTTP/2 request unexpectedly used HTTP/1");
         assert!(error.is_user());
         let error_connection = error
             .connection_metadata()
             .expect("the mismatch error omitted connection metadata");
         let captured_connection = capture
             .get()
             .expect("the mismatch request did not capture its connection");
         assert_eq!(
             error_connection.connection_id(),
             captured_connection.connection_id()
         );
-1179
         let response = request(&pool, partition, server.uri("/h1")).await;
         assert_eq!(
             hyper::body::Bytes::from_static(b"ok"),
             consume(response).await
         );
         assert_eq!(
 ,
             server.accepted(),
             "the rejected HTTP/2 request discarded a reusable HTTP/1 connection"
         );
-1190
+    }
-1191
     #[tokio::test]
     async fn captured_metadata_poisons_only_selected_generation() {
         let server = TestServer::start().await;
         let pool = ConnectionPool::builder().build_http().unwrap();
         let partition = anonymous_partition(&pool);
         let capture = CaptureSmithyConnection::new();
         let mut first = Request::get(server.uri("/first"))
             .body(SdkBody::empty())
             .unwrap();
         first.extensions_mut().insert(capture.clone());
         let response = pool
             .send_request(partition.clone(), first, None)
             .send_request(partition.clone(), first, RequestOptions::default())
             .await
             .unwrap();
         consume(response).await;
-1207
         let metadata = capture.get().expect("connection metadata was captured");
         assert!(metadata.connection_id().is_some());
         metadata.poison();
-1211
         consume(request(&pool, partition.clone(), server.uri("/second")).await).await;
         assert_eq!(2, server.accepted());
-1214
         metadata.poison();
         consume(request(&pool, partition, server.uri("/third")).await).await;
         assert_eq!(
 ,
             server.accepted(),
             "stale metadata poisoned the newer connection generation"
         );
-1222
+    }
-1223
     #[tokio::test]
     async fn connection_id_exhaustion_is_an_internal_error() {
         let server = TestServer::start().await;
         let pool = ConnectionPool::builder().build_http().unwrap();
         pool.inner
             .next_connection_id
             .store(u64::MAX, Ordering::Relaxed);
         let partition = anonymous_partition(&pool);
         let request = Request::get(server.uri("/id-exhausted"))
             .body(SdkBody::empty())
             .unwrap();
-1235
         let error = pool
             .send_request(partition, request, None)
             .send_request(partition, request, RequestOptions::default())
             .await
             .expect_err("exhausted connection identifiers unexpectedly established a connection");
         assert!(error.is_other());
         assert_eq!(
             "connection identifier space exhausted",
             std::error::Error::source(&error).unwrap().to_string()
         );
-1245
+    }
-1246
     #[tokio::test]
     async fn explicit_clients_resolve_declared_partitions() {
         let pool = ConnectionPool::builder()
             .partitions([Partition::new(
                 PartitionId::from_index(7),
                 TokioDriverSpawner::current(),
             )])
             .build_http()
             .unwrap();
-1256
         assert!(Client::new(&pool).is_err());
         assert!(Client::from_partition(&pool, PartitionId::from_index(7)).is_ok());
         let error = Client::from_partition(&pool, PartitionId::from_index(8)).unwrap_err();
         assert_eq!(Some(PartitionId::from_index(8)), error.partition());
-1261
+    }
-1262
     #[test]
     fn anonymous_partition_outside_tokio_is_a_user_error() {
         let pool = ConnectionPool::builder()
             .idle_timeout(None)
             .build_http()
             .unwrap();
         let partition = anonymous_partition(&pool);
         let request = Request::get("http://example.com/")
             .body(SdkBody::empty())
             .unwrap();
         let mut send = Box::pin(pool.send_request(partition, request, None));
         let mut send = Box::pin(pool.send_request(partition, request, RequestOptions::default()));
         let waker = std::task::Waker::noop();
         let mut context = Context::from_waker(waker);
-1276
         let Poll::Ready(Err(error)) = send.as_mut().poll(&mut context) else {
             panic!("missing anonymous runtime did not fail on first poll");
         };
         assert!(error.is_user());
         assert_eq!(
             "the anonymous connection-pool partition requires an active Tokio runtime on first use",
             std::error::Error::source(&error).unwrap().to_string()

↧ Expand context

         );
-1285
+    }
-1286
     #[tokio::test]
     async fn registry_shutdown_stops_partition_maintenance() {
         let active = StdArc::new(AtomicUsize::new(0));
         let pool = ConnectionPool::builder()
             .partitions([Partition::new(
                 PartitionId::from_index(7),
                 TrackingSpawner {
                     active: active.clone(),
                 },
             )])
             .build_http()
             .unwrap();
         let partition = pool
             .inner
             .registry
             .partition(PartitionId::from_index(7))
             .unwrap();

@@ -1414,1406 +1474,1466 @@

↥ Expand context

                 Partition::new(mismatched_id, TokioDriverSpawner::current())
                     .interface("synthetic-interface-b"),
             ])
             .connection_reuse_scope(ConnectionReuseScope::NetworkInterface)
             .max_connections_per_host(1)
             // Injected connectors deliberately ignore OS interface placement.
             .build_with_connector(ExtraConnector)
             .unwrap();
         let first = pool.inner.registry.partition(first_id).unwrap();
         let matching = pool.inner.registry.partition(matching_id).unwrap();
         let mismatched = pool.inner.registry.partition(mismatched_id).unwrap();
         let first_uri = server.uri("/first-interface");
-1418
         consume(request(&pool, first.clone(), first_uri.clone()).await).await;
         let first_connection = pool
             .inner
             .registry
             .resolve_cell(&first, &first_uri)
             .unwrap()
             .only_h1_connection_for_test();

         consume(request(&pool, matching, server.uri("/matching-interface")).await).await;
         assert_eq!(
 ,
             server.accepted(),
             "matching interface groups did not share the existing connection"
         );
-1432
         consume(request(&pool, mismatched, server.uri("/mismatched-interface")).await).await;
         assert_eq!(
             Some(CloseReason::Reclaimed),
             first_connection.snapshot().close_reason
             first_connection.probe().close_reason
         );
         assert_eq!(
 ,
             server.accepted(),
             "mismatched interface groups did not reclaim bounded capacity"
         );
-1443
+    }
-1444
     #[tokio::test]
     async fn smithy_client_boundary_sends_through_the_shared_pool() {

↧ Expand context

         let server = TestServer::start().await;
         let pool = ConnectionPool::builder().build_http().unwrap();
         let client = Client::new(&pool).unwrap();
         let components = RuntimeComponentsBuilder::for_tests().build().unwrap();
         let connector =
             client.http_connector(&HttpConnectorSettings::builder().build(), &components);
-1453
         let mut response = connector
             .call(HttpRequest::get(server.uri("/smithy").to_string()).unwrap())
             .await
             .unwrap();
         let body = response.take_body().collect().await.unwrap().to_bytes();
         assert_eq!(hyper::body::Bytes::from_static(b"ok"), body);
-1460
+    }
-1461
     #[tokio::test]
     async fn connect_timeout_covers_the_transport_operation() {
         let pool = super::super::super::builder::Builder::default()
             .build_with_connector(NeverConnects)
             .unwrap();

@@ -1542,1534 +1631,1627 @@

↥ Expand context

                     });
                 } else {
                     tokio::spawn(serve_connection(stream));
-1537
+                }
-1538
+            }
         });
-1540
         let pool = ConnectionPool::builder()
             .max_connections_per_host(1)
             .build_http()
             .unwrap();
         let partition = anonymous_partition(&pool);
         let upgrade_uri: Uri = format!("{endpoint}/upgrade").parse().unwrap();
         let mut upgrade_request = Request::builder()
             .method(method.clone())
             .uri(upgrade_uri.clone())
             .body(SdkBody::empty())
             .unwrap();
         if method != Method::CONNECT {
             upgrade_request

                 .headers_mut()
                 .insert(http_1x::header::CONNECTION, "upgrade".parse().unwrap());
             upgrade_request
                 .headers_mut()
                 .insert(http_1x::header::UPGRADE, "test".parse().unwrap());
-1559
+        }
         let request_pool = pool.clone();
         let request_partition = partition.clone();
         let response = tokio::spawn(async move {
             request_pool
                 .send_request(request_partition, upgrade_request, None)
                 .send_request(
                     request_partition,
                     upgrade_request,
                     RequestOptions::default(),
-+
+                )
                 .await
                 .unwrap()
         });
         request_received.notified().await;
         let cell = pool
             .inner
             .registry
             .resolve_cell(&partition, &upgrade_uri)
             .unwrap();
         let connection = cell.only_h1_connection_for_test();
         release_response.notify_one();
         let mut response = response.await.unwrap();
         assert_eq!(expected_status, response.status());
-1582
         let upgraded = tokio::time::timeout(
             std::time::Duration::from_secs(1),
             hyper::upgrade::on(&mut response),
-1586
+        )
         .await
         .expect("upgrade timed out")
         .expect("upgrade failed");
         let mut upgraded = hyper_util::rt::TokioIo::new(upgraded);
         let mut greeting = [0_u8; 5];
         upgraded.read_exact(&mut greeting).await.unwrap();
         assert_eq!(b"hello", &greeting);
-1594
         assert_eq!(
             Some(CloseReason::Upgraded),
             connection.snapshot().close_reason,
             connection.probe().close_reason,
             "the upgraded HTTP/1 sender returned to pool policy"
         );
         assert_eq!(1, cell.admission().unwrap().available_capacity_for_test());
-1601
         let second_uri: Uri = format!("{endpoint}/after").parse().unwrap();
         let second = tokio::time::timeout(
             std::time::Duration::from_secs(1),
             request(&pool, partition, second_uri),
-1606
+        )
         .await

↧ Expand context

         .expect("bounded capacity was not released at upgrade logical close");
         consume(second).await;
         assert_eq!(2, accepted.load(Ordering::SeqCst));
-1611
         drop(upgraded);
         drop(response);
         server.abort();
-1615
+    }
-1616
     #[tokio::test]
     async fn switching_protocols_upgrade_releases_capacity_and_transfers_io() {
         assert_h1_upgrade_releases_capacity(
             Method::GET,
             b"HTTP/1.1 101 Switching Protocols\r\n\
               connection: upgrade\r\n\
               upgrade: test\r\n\r\nhello",
             http_1x::StatusCode::SWITCHING_PROTOCOLS,
-1625
+        )
         .await;
-1627
+    }

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

@@ -1,1 +642,639 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! HTTP/2 dispatch and two-ended request completion.
 //!
 //! [`H2Activation`] arrives with a prospective generation lease and a transient
 //! [`H2Activation`] arrives with prospective request authority and a transient
 //! sender cloned from the connection-owning cell. Dispatch first checks sender
 //! and logical-connection state, then polls Hyper once. A request returned from
 //! that poll was not accepted and may re-enter protocol acquisition.
 //!
 //! After Hyper accepts the request, the activation becomes an accepted lease.
 //! [`H2RequestBody`] owns its send endpoint and [`H2ResponseBody`] owns its
 //! receive endpoint. Either may finish first; the generation request count is
 //! After Hyper accepts the request, the activation becomes an accepted claim.
 //! [`H2RequestBody`] owns the upload guard and [`H2ResponseBody`] owns the
 //! response guard. Either may finish first; the generation request count is
 //! released only after both finish. A returned request re-arms the existing
 //! body wrapper rather than nesting wrappers around the original body.
-18
 use super::super::cell::h2::{H2Activation, H2CloseHandle, H2DispatchParts, H2LeaseEndpoint};
 use super::super::connection::ConnectionState;
 use super::super::ConnectionPool;
 use super::super::cell::h2::{
     H2Activation, H2CloseHandle, H2DispatchParts, H2ResponseGuard, H2UploadGuard,
 };
 use super::super::connection::{CloseReason, ConnectionState};
 use super::{AcquisitionContext, H1HostHeaderInserted};
 use crate::client::downcast_error;
 use crate::sync::{Arc, Mutex};
 use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use http_1x::{Request, Response};
 use hyper::body::Body;
 use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::task::{Context, Poll};
-34
 /// Result of one checked HTTP/2 dispatch attempt.
 #[allow(
     clippy::large_enum_variant,
     reason = "boxing the successful response would allocate on every HTTP/2 request"
 )]
 pub(super) enum H2DispatchResult {
 pub(super) enum H2DispatchOutcome {
     /// Hyper accepted the request and produced a guarded response.
     Response(Response<SdkBody>),
     /// Hyper did not accept the request, so it may re-enter acquisition.
     Reacquire(Box<H2Reacquisition>),
     NotAccepted(Box<H2UnacceptedRequest>),
-45
+}
-46
 /// Request and terminal fallback retained for one replacement selection.
 pub(super) struct H2Reacquisition {
 pub(super) struct H2UnacceptedRequest {
     /// Original request Hyper did not accept.
     request: Request<SdkBody>,
     /// Error returned if the request exhausts its replacement budget.
     error: ConnectorError,
     terminal_error: ConnectorError,
-53
+}
-54
 impl H2Reacquisition {
 impl H2UnacceptedRequest {
     /// Returns the original request and its terminal fallback.
     pub(super) fn into_parts(self: Box<Self>) -> (Request<SdkBody>, ConnectorError) {
         let Self { request, error } = *self;
         (request, error)
         let Self {
             request,
             terminal_error,
         } = *self;
         (request, terminal_error)
-63
+    }
-64
+}
-65
 /// State retained after Hyper accepts one request.
 struct H2AcceptedDispatch {
     /// Connection retained through response creation.
     connection: Arc<ConnectionState>,
     /// Generation close authority used for connection-wide failures.
     close: H2CloseHandle,
     /// Metadata captured before the request moved into Hyper.
     captured_metadata: Option<aws_smithy_runtime_api::client::connection::ConnectionMetadata>,
     /// Receive endpoint transferred to the guarded response body.
     receive_endpoint: H2LeaseEndpoint,
     /// Response guard transferred to the guarded response body.
     response_guard: H2ResponseGuard,
     /// Whether the generation accepted an earlier request.
     reused: bool,
-78
+}
-79
 impl ConnectionPool {
     /// Dispatches one request through a prospective H2 generation lease.
     ///
     /// Pool-side staleness reacquires before Hyper sees the request. Once
     /// Hyper returns an envelope, only a reused generation may reacquire.
     pub(super) async fn dispatch_h2(
         &self,
 /// Dispatches one request through prospective H2 request authority.
 ///
 /// Pool-side staleness reacquires before Hyper sees the request. Once Hyper
 /// returns an envelope, only a reused generation may reacquire.
 pub(super) async fn dispatch(
     context: &AcquisitionContext,
     mut request: Request<SdkBody>,
     mut activation: H2Activation,
     ) -> Result<H2DispatchResult, ConnectorError> {
 ) -> Result<H2DispatchOutcome, ConnectorError> {
     prepare_h2_request(&mut request, &context.absolute_uri);
     let connection = activation.connection().clone();
     let reused = activation.is_reused();
     let close = activation.close_handle();
     let captured_metadata = request
         .extensions()
         .get::<CaptureSmithyConnection>()
         .cloned()
         .map(|capture| {
             let metadata = connection.info().h2_metadata(close.clone());
             let captured = metadata.clone();
             capture.set_connection_retriever(move || Some(captured.clone()));
             metadata
         });
-103
     let H2DispatchParts {
         mut sender,
             send_endpoint,
             receive_endpoint,
         upload,
         response,
     } = activation.take_dispatch_parts();
     if sender.is_closed() {
             close.close(super::super::connection::CloseReason::ProtocolClosed);
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
             return finish_unaccepted_request(
         close.close(CloseReason::ProtocolClosed);
         let metadata = captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
         return resolve_unaccepted_request(
             request,
             reused,
             UnacceptedStage::BeforeHyper,
             ConnectorError::other(H2ConnectionClosedBeforeDispatch.into(), None)
                 .with_connection(metadata),
         );
-119
+    }
     let Some(dispatch) = ConnectionState::try_commit_dispatch(&connection) else {
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
             return finish_unaccepted_request(
         let metadata = captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
         return resolve_unaccepted_request(
             request,
             reused,
             UnacceptedStage::BeforeHyper,
             ConnectorError::other(H2ConnectionClosedBeforeDispatch.into(), None)
                 .with_connection(metadata),
         );
     };
-130
         let body = H2RequestBodyHandle::arm(&mut request, send_endpoint);
     let body = H2RequestBodyHandle::arm(&mut request, upload);
-132
     let mut send = Box::pin(sender.hyper_mut().try_send_request(request));
     let first = poll_fn(|cx| Poll::Ready(send.as_mut().poll(cx))).await;
-135
     match first {
         Poll::Ready(Err(mut error)) if error.message().is_some() => {
             let returned = error
                 .take_message()
                 .expect("checked returned request disappeared");
             body.clear();
                 drop(receive_endpoint);
             drop(response);
             drop(dispatch);
             drop(activation);
                 close.close(super::super::connection::CloseReason::ProtocolClosed);
             close.close(CloseReason::ProtocolClosed);
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
                 finish_unaccepted_request(
             resolve_unaccepted_request(
                 returned,
                 reused,
                 UnacceptedStage::ReturnedByHyper,
                     super::super::super::downcast_error(Box::new(error.into_error()))
                         .with_connection(metadata),
                 downcast_error(Box::new(error.into_error())).with_connection(metadata),
-153
+            )
-154
+        }
         Poll::Ready(result) => {
             activation.accept(dispatch);
             let sender_closed = sender.is_closed();
                 self.finish_h2_send(
             resolve_h2_send(
                 sender_closed,
                 result,
                 H2AcceptedDispatch {
                     connection,
                     close,
                     captured_metadata,
                         receive_endpoint,
                     response_guard: response,
                     reused,
                 },
-168
+            )
-169
+        }
         Poll::Pending => {
             activation.accept(dispatch);
             let result = send.await;
             let sender_closed = sender.is_closed();
                 self.finish_h2_send(
             resolve_h2_send(
                 sender_closed,
                 result,
                 H2AcceptedDispatch {
                     connection,
                     close,
                     captured_metadata,
                         receive_endpoint,
                     response_guard: response,
                     reused,
                 },
-184
+            )
-185
+        }
-186
+    }
--
+    }
-+
+}
-188
     #[allow(
 #[allow(
     clippy::result_large_err,
     reason = "ConnectorError preserves SDK classification and connection metadata"
     )]
     fn finish_h2_send(
         &self,
 )]
 fn resolve_h2_send(
     sender_closed: bool,
     result: Result<
         Response<hyper::body::Incoming>,
         hyper::client::conn::TrySendError<Request<SdkBody>>,
     >,
     accepted: H2AcceptedDispatch,
     ) -> Result<H2DispatchResult, ConnectorError> {
 ) -> Result<H2DispatchOutcome, ConnectorError> {
     let H2AcceptedDispatch {
         connection,
         close,
         captured_metadata,
             receive_endpoint,
         response_guard,
         reused,
     } = accepted;
     match result {
         Ok(mut response) => {
             connection
                 .info()
                 .apply_connector_extras(response.extensions_mut());
             let (parts, body) = response.into_parts();
                 let body = H2ResponseBody::new(body, receive_endpoint);
                 Ok(H2DispatchResult::Response(Response::from_parts(
             let body = H2ResponseBody::new(body, response_guard);
             Ok(H2DispatchOutcome::Response(Response::from_parts(
                 parts,
                 SdkBody::from_body_1_x(body),
             )))
-219
+        }
         Err(mut error) => {
             // Hyper may return a queued envelope when its connection task
             // drops after the first poll. The returned request is the
             // authority for reacquisition; an error without it is terminal.
             if let Some(request) = error.take_message() {
                 if let Some(body) = request.extensions().get::<H2RequestBodyHandle>() {
                     body.clear();
-227
+                }
                     drop(receive_endpoint);
                     close.close(super::super::connection::CloseReason::ProtocolClosed);
                 drop(response_guard);
                 close.close(CloseReason::ProtocolClosed);
                 let metadata =
                     captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
                     return finish_unaccepted_request(
                 return resolve_unaccepted_request(
                     request,
                     reused,
                     UnacceptedStage::ReturnedByHyper,
                         super::super::super::downcast_error(Box::new(error.into_error()))
                             .with_connection(metadata),
                     downcast_error(Box::new(error.into_error())).with_connection(metadata),
                 );
-238
+            }
             if sender_closed {
                     close.close(super::super::connection::CloseReason::ProtocolClosed);
                 close.close(CloseReason::ProtocolClosed);
-241
+            }
                 drop(receive_endpoint);
             drop(response_guard);
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
                 Err(
                     super::super::super::downcast_error(Box::new(error.into_error()))
                         .with_connection(metadata),
--
+                )
--
+            }
             Err(downcast_error(Box::new(error.into_error())).with_connection(metadata))
-246
+        }
-247
+    }
-248
+}
-249
 /// Restores the absolute URI and removes only a pool-synthesized H1 `Host`.
 fn prepare_h2_request(request: &mut Request<SdkBody>, absolute_uri: &http_1x::Uri) {
     *request.uri_mut() = absolute_uri.clone();
     if request
         .extensions_mut()
         .remove::<H1HostHeaderInserted>()
         .is_some()
-257
+    {
         request.headers_mut().remove(http_1x::header::HOST);
-259
+    }
-260
+}
-261
 /// Point at which the pool learned that Hyper had not accepted the request.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum UnacceptedStage {
     /// The pool rejected a stale generation before calling Hyper.
     BeforeHyper,
     /// Hyper returned the original request envelope.
     ReturnedByHyper,
-269
+}
-270
 /// Applies retry authority to a request that Hyper did not accept.
 ///
 /// Pool-side checks always reacquire because no protocol code observed the
 /// request. A returned Hyper envelope reacquires only after prior successful
 /// use proves that replacing a stale pooled generation is appropriate.
 #[allow(
     clippy::result_large_err,
     reason = "ConnectorError preserves SDK classification and connection metadata"
 )]
 fn finish_unaccepted_request(
 fn resolve_unaccepted_request(
     request: Request<SdkBody>,
     reused: bool,
     stage: UnacceptedStage,
     error: ConnectorError,
 ) -> Result<H2DispatchResult, ConnectorError> {
 ) -> Result<H2DispatchOutcome, ConnectorError> {
     if stage == UnacceptedStage::BeforeHyper || reused {
         Ok(H2DispatchResult::Reacquire(Box::new(H2Reacquisition {
         Ok(H2DispatchOutcome::NotAccepted(Box::new(
             H2UnacceptedRequest {
                 request,
             error,
         })))
                 terminal_error: error,
             },
         )))
     } else {
         Err(error)
-295
+    }
-296
+}
-297
 /// A fresh HTTP/2 generation closed before accepting its first request.
 #[derive(Debug)]
 struct H2ConnectionClosedBeforeDispatch;
-301
 impl std::fmt::Display for H2ConnectionClosedBeforeDispatch {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("the fresh HTTP/2 connection closed before request dispatch")
-305
+    }
-306
+}
-307
 impl std::error::Error for H2ConnectionClosedBeforeDispatch {}
-309
 /// Handle retained in a request extension after its body is wrapped once.
 #[derive(Clone)]
 struct H2RequestBodyHandle {
     /// Re-armable send endpoint shared with the wrapped request body.
     slot: Arc<Mutex<Option<H2LeaseEndpoint>>>,
     /// Re-armable upload guard shared with the wrapped request body.
     slot: Arc<Mutex<Option<H2UploadGuard>>>,
-315
+}
-316
 impl H2RequestBodyHandle {
     /// Wraps an unwrapped body and arms the current send endpoint.
     fn arm(request: &mut Request<SdkBody>, endpoint: H2LeaseEndpoint) -> Self {
     /// Wraps an unwrapped body and arms the current upload guard.
     fn arm(request: &mut Request<SdkBody>, upload: H2UploadGuard) -> Self {
         let is_end_stream = request.body().is_end_stream();
         let handle = request
             .extensions()
             .get::<Self>()
             .cloned()
             .unwrap_or_else(|| {
                 let handle = Self {
                     slot: Arc::new(Mutex::new(None)),
                 };
                 let body = std::mem::replace(request.body_mut(), SdkBody::taken());
                 *request.body_mut() = SdkBody::from_body_1_x(H2RequestBody {
                     inner: body,
                     slot: handle.slot.clone(),
                 });
                 request.extensions_mut().insert(handle.clone());
                 handle
             });
         arm_send_endpoint(&handle.slot, endpoint);
         arm_upload_guard(&handle.slot, upload);
         if is_end_stream {
             finish_send_endpoint(&handle.slot);
             finish_upload(&handle.slot);
-340
+        }
         handle
-342
+    }
-343
     /// Disarms an endpoint after Hyper returns the request unaccepted.
     /// Disarms the upload guard after Hyper returns the request unaccepted.
     fn clear(&self) {
         finish_send_endpoint(&self.slot);
         finish_upload(&self.slot);
-347
+    }
-348
+}
-349
 /// Request body whose endpoint can be re-armed after certified non-acceptance.
 /// Request body whose upload guard can be re-armed after non-acceptance.
 struct H2RequestBody {
     /// Original SDK body wrapped exactly once.
     inner: SdkBody,
     /// Endpoint replaced when Hyper returns an unaccepted request.
     slot: Arc<Mutex<Option<H2LeaseEndpoint>>>,
     /// Guard replaced when Hyper returns an unaccepted request.
     slot: Arc<Mutex<Option<H2UploadGuard>>>,
-356
+}
-357
 impl Body for H2RequestBody {
     type Data = <SdkBody as Body>::Data;
     type Error = <SdkBody as Body>::Error;
-361
     fn poll_frame(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
         let result = Pin::new(&mut self.inner).poll_frame(cx);
         if matches!(result, Poll::Ready(None) | Poll::Ready(Some(Err(_)))) {
             finish_send_endpoint(&self.slot);
             finish_upload(&self.slot);
-369
+        }
         result
-371
+    }
-372
     fn is_end_stream(&self) -> bool {
         self.inner.is_end_stream()
-375
+    }
-376
     fn size_hint(&self) -> hyper::body::SizeHint {
         self.inner.size_hint()
-379
+    }
-380
+}
-381
 impl Drop for H2RequestBody {
     fn drop(&mut self) {
         finish_send_endpoint(&self.slot);
         finish_upload(&self.slot);
-385
+    }
-386
+}
-387
 /// Response body that owns the receive endpoint through stream completion.
 /// Response body that owns the response guard through stream completion.
 struct H2ResponseBody {
     /// Hyper response stream.
     inner: hyper::body::Incoming,
     /// Receive endpoint completed on terminal frame, error, or drop.
     receive: Option<H2LeaseEndpoint>,
     /// Response guard finished on terminal frame, error, or drop.
     response: Option<H2ResponseGuard>,
-394
+}
-395
 impl H2ResponseBody {
     /// Wraps a response and completes an endpoint already at end stream.
     fn new(inner: hyper::body::Incoming, receive: H2LeaseEndpoint) -> Self {
     /// Wraps a response and finishes a guard already at end stream.
     fn new(inner: hyper::body::Incoming, response: H2ResponseGuard) -> Self {
         let is_end_stream = inner.is_end_stream();
         Self {
             inner,
             receive: retain_receive_endpoint(is_end_stream, receive),
             response: retain_response_guard(is_end_stream, response),
-403
+        }
-404
+    }
-405
+}
-406
 /// Retains a receive endpoint only while response frames may remain.
 fn retain_receive_endpoint(
 /// Retains a response guard only while response frames may remain.
 fn retain_response_guard(
     is_end_stream: bool,
     receive: H2LeaseEndpoint,
 ) -> Option<H2LeaseEndpoint> {
     response: H2ResponseGuard,
 ) -> Option<H2ResponseGuard> {
     if is_end_stream {
         receive.complete();
         response.finish();
         None
     } else {
         Some(receive)
         Some(response)
-417
+    }
-418
+}
-419
 impl Body for H2ResponseBody {
     type Data = <hyper::body::Incoming as Body>::Data;
     type Error = <hyper::body::Incoming as Body>::Error;
-423
     fn poll_frame(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
         let result = Pin::new(&mut self.inner).poll_frame(cx);
         if matches!(result, Poll::Ready(None) | Poll::Ready(Some(Err(_)))) {
             if let Some(endpoint) = self.receive.take() {
                 endpoint.complete();
             if let Some(response) = self.response.take() {
                 response.finish();
-432
+            }
-433
+        }
         result
-435
+    }
-436
     fn is_end_stream(&self) -> bool {
         self.inner.is_end_stream()
-439
+    }
-440
     fn size_hint(&self) -> hyper::body::SizeHint {
         self.inner.size_hint()
-443
+    }
-444
+}
-445
 fn arm_send_endpoint(slot: &Arc<Mutex<Option<H2LeaseEndpoint>>>, endpoint: H2LeaseEndpoint) {
     let previous = slot.lock().replace(endpoint);
 fn arm_upload_guard(slot: &Arc<Mutex<Option<H2UploadGuard>>>, upload: H2UploadGuard) {
     let previous = slot.lock().replace(upload);
     drop(previous);
-449
+}
-450
 /// Terminates the send endpoint after upload completion, error, or rejection.
 fn finish_send_endpoint(slot: &Arc<Mutex<Option<H2LeaseEndpoint>>>) {
     let endpoint = slot.lock().take();
     if let Some(endpoint) = endpoint {
         endpoint.complete();
 /// Finishes the upload guard after upload completion, error, or rejection.
 fn finish_upload(slot: &Arc<Mutex<Option<H2UploadGuard>>>) {
     let upload = slot.lock().take();
     if let Some(upload) = upload {
         upload.finish();
-456
+    }
-457
+}
-458
 #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
 mod tests {
     use super::*;
     use crate::client::pool::cell::OriginCell;
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::{EligibilityGroup, PartitionId};
     use crate::sync::Arc;
     use http_1x::uri::Scheme;
     use http_body_util::BodyExt;
-468
     /// Request body that fails on its first frame.
     struct FailingBody;
-471
     impl Body for FailingBody {
         type Data = hyper::body::Bytes;
         type Error = std::io::Error;
-475
         fn poll_frame(
             self: Pin<&mut Self>,
             _cx: &mut Context<'_>,
         ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
             Poll::Ready(Some(Err(std::io::Error::other("request body failed"))))
-481
+        }
-482
+    }
-483
     fn cell() -> Arc<OriginCell> {
         Arc::new(OriginCell::new(
             PartitionId::from_index(1),
             OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
             EligibilityGroup::Pool,
             None,
             None,
         ))
-492
+    }
-493
     #[tokio::test]
     async fn returned_request_rearms_one_body_wrapper() {
         let cell = cell();
         let mut request = Request::post("https://example.com/upload")
             .body(SdkBody::from("payload"))
             .unwrap();
-500
         let (first_endpoint, first_probe) = H2LeaseEndpoint::send_for_test(&cell);
         let first = H2RequestBodyHandle::arm(&mut request, first_endpoint);
         let (first_upload, first_probe) = H2UploadGuard::for_test(&cell);
         let first = H2RequestBodyHandle::arm(&mut request, first_upload);
         let first_slot = first.slot.clone();
         assert!(!first_probe.send_complete());
         assert!(!first_probe.upload_finished());
-505
         let (second_endpoint, second_probe) = H2LeaseEndpoint::send_for_test(&cell);
         let second = H2RequestBodyHandle::arm(&mut request, second_endpoint);
         let (second_upload, second_probe) = H2UploadGuard::for_test(&cell);
         let second = H2RequestBodyHandle::arm(&mut request, second_upload);
         assert!(
             Arc::ptr_eq(&first_slot, &second.slot),
             "rearming replaced the request-body wrapper"
         );
         assert!(
             first_probe.send_complete(),
             "rearming did not cancel the prior prospective endpoint"
             first_probe.upload_finished(),
             "rearming did not cancel the prior prospective upload guard"
         );
         assert!(!second_probe.send_complete());
         assert!(!second_probe.upload_finished());
-517
         second.clear();
         assert!(
             second_probe.send_complete(),
             "returned request retained its rejected send endpoint"
             second_probe.upload_finished(),
             "returned request retained its rejected upload guard"
         );
         let body = request
             .into_body()
             .collect()
             .await
             .expect("wrapped request body failed")
             .to_bytes();
         assert_eq!(hyper::body::Bytes::from_static(b"payload"), body);
-530
+    }
-531
     #[test]
     fn empty_request_body_completes_its_send_endpoint_when_armed() {
     fn empty_request_body_finishes_its_upload_guard_when_armed() {
         let cell = cell();
         let mut request = Request::new(SdkBody::empty());
         let (endpoint, probe) = H2LeaseEndpoint::send_for_test(&cell);
         let (upload, probe) = H2UploadGuard::for_test(&cell);
-537
         H2RequestBodyHandle::arm(&mut request, endpoint);
         H2RequestBodyHandle::arm(&mut request, upload);
-539
         assert!(probe.send_complete());
         assert!(probe.upload_finished());
-541
+    }
-+
     #[test]
     fn dropping_request_body_completes_its_send_endpoint() {
     fn dropping_request_body_finishes_its_upload_guard() {
         let cell = cell();
         let mut request = Request::new(SdkBody::from("payload"));
         let (endpoint, probe) = H2LeaseEndpoint::send_for_test(&cell);
         let (upload, probe) = H2UploadGuard::for_test(&cell);
-548
         H2RequestBodyHandle::arm(&mut request, endpoint);
         H2RequestBodyHandle::arm(&mut request, upload);
         let body = std::mem::replace(request.body_mut(), SdkBody::empty());
         drop(body);
-552
         assert!(probe.send_complete());
         assert!(probe.upload_finished());
-554
+    }
-555
     #[tokio::test]
     async fn request_body_error_completes_its_send_endpoint() {
     async fn request_body_error_finishes_its_upload_guard() {
         let cell = cell();
         let mut request = Request::new(SdkBody::from_body_1_x(FailingBody));
         let (endpoint, probe) = H2LeaseEndpoint::send_for_test(&cell);
         let (upload, probe) = H2UploadGuard::for_test(&cell);
-561
         H2RequestBodyHandle::arm(&mut request, endpoint);
         H2RequestBodyHandle::arm(&mut request, upload);
         let frame = request
             .body_mut()
             .frame()
             .await
             .expect("failing body omitted its error frame");
-568
         assert!(frame.is_err());
         assert!(probe.send_complete());
         assert!(probe.upload_finished());
-571
+    }
-572
     #[test]
     fn bodyless_response_completes_its_receive_endpoint_without_drop() {
     fn bodyless_response_finishes_its_response_guard_without_drop() {
         let cell = cell();
         let (endpoint, probe) = H2LeaseEndpoint::receive_for_test(&cell);
         let (response, probe) = H2ResponseGuard::for_test(&cell);
-577
         let retained = retain_receive_endpoint(true, endpoint);
         let retained = retain_response_guard(true, response);
-579
         assert!(retained.is_none());
         assert!(probe.receive_complete());
         assert!(probe.response_finished());
-582
+    }
-583
     #[test]
     fn h2_preparation_restores_uri_and_removes_only_synthesized_host() {
         let absolute: http_1x::Uri = "https://example.com/resource".parse().unwrap();
         let mut request = Request::get("/resource").body(SdkBody::empty()).unwrap();
         request
             .headers_mut()
             .insert(http_1x::header::HOST, "example.com".parse().unwrap());
         request.extensions_mut().insert(H1HostHeaderInserted);
-592
         prepare_h2_request(&mut request, &absolute);
-594
         assert_eq!(&absolute, request.uri());
         assert!(!request.headers().contains_key(http_1x::header::HOST));
         assert!(request.extensions().get::<H1HostHeaderInserted>().is_none());
-598
         let mut user_host = Request::get("/resource").body(SdkBody::empty()).unwrap();
         user_host
             .headers_mut()
             .insert(http_1x::header::HOST, "signed.example".parse().unwrap());
         prepare_h2_request(&mut user_host, &absolute);
         assert_eq!("signed.example", user_host.headers()[http_1x::header::HOST]);
-605
+    }
-606
     #[test]
     fn retry_authority_distinguishes_pool_checks_from_hyper_returns() {
         let checked_before_hyper = finish_unaccepted_request(
         let checked_before_hyper = resolve_unaccepted_request(
             Request::new(SdkBody::empty()),
             false,
             UnacceptedStage::BeforeHyper,
             ConnectorError::user("pre-Hyper failure".into()),
         );
         assert!(matches!(
             checked_before_hyper,
             Ok(H2DispatchResult::Reacquire(_))
             Ok(H2DispatchOutcome::NotAccepted(_))
         ));
-619
         let fresh_hyper_return = finish_unaccepted_request(
         let fresh_hyper_return = resolve_unaccepted_request(
             Request::new(SdkBody::empty()),
             false,
             UnacceptedStage::ReturnedByHyper,
             ConnectorError::user("fresh Hyper failure".into()),
         );
         assert!(fresh_hyper_return.is_err());
-627
         let reused_hyper_return = finish_unaccepted_request(
         let reused_hyper_return = resolve_unaccepted_request(
             Request::new(SdkBody::empty()),
             true,
             UnacceptedStage::ReturnedByHyper,
             ConnectorError::user("reused failure".into()),
         );
         assert!(matches!(
             reused_hyper_return,
             Ok(H2DispatchResult::Reacquire(_))
             Ok(H2DispatchOutcome::NotAccepted(_))
         ));
-638
+    }
-639
+}

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

@@ -1,1 +380,162 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Transport construction below HTTP protocol establishment.
 //! Transport connection followed by HTTP protocol establishment.
 //!
 //! [`transport`] creates connected I/O and reports the negotiated protocol.
 //! This module then routes that transport through the matching Hyper handshake,
 //! creates the protocol-neutral connection identity, and returns or transfers
 //! completion ownership for the launching waiter.
-12
 mod h1;
 mod h2;
 mod transport;
-16
 #[cfg(any(feature = "__rustls", feature = "s2n-tls"))]
 pub(super) use transport::from_cached_interface_connector;
 #[cfg(any(
     all(feature = "test-util", aws_sdk_unstable),
     all(test, feature = "rt-tokio")
 ))]
 pub(super) use transport::from_connector;
 pub(super) use transport::{from_interface_connector, TransportFactory, TransportTimeout};
-+
 use self::transport::TransportConnectContext;
 use super::admission::ProtocolRequirement;
 use super::cell::{AcquisitionResult, EstablishmentPermit, WaiterId};
 use super::cell::{AcquisitionOutcome, EstablishmentPermit, WaiterId};
 use super::dispatch::AcquisitionContext;
 use super::registry::PartitionState;
 use super::PoolInner;
 use crate::client::connect::{AsyncConn, BoxConn};
 use crate::client::timeout::{self, TimeoutKind};
 use aws_smithy_async::rt::sleep::SharedAsyncSleep;
 use aws_smithy_runtime_api::box_error::BoxError;
 use crate::client::downcast_error;
 use aws_smithy_runtime_api::client::connection::ConnectionId;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use http_1x::Uri;
 #[cfg(any(feature = "__rustls", feature = "s2n-tls"))]
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt;
 use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::sync::atomic::Ordering;
 use std::sync::Arc as StdArc;
 use std::time::Duration;
 use tower::Service;
--
 /// Timeout policy for one transport connection operation.
 ///
 /// Connector readiness completes before this timeout starts. Keeping the
 /// duration and timer together lets request acquisition pass one complete
 /// timeout value into transport construction.
 #[derive(Clone, Debug)]
 pub(super) struct TransportTimeout {
     /// Maximum duration of the connector's connection future.
     duration: Duration,
     /// Runtime timer used to enforce the deadline.
     sleep: SharedAsyncSleep,
--
+}
--
 impl TransportTimeout {
     /// Creates a timeout enforced by the supplied runtime timer.
     pub(super) fn new(duration: Duration, sleep: SharedAsyncSleep) -> Self {
         Self { duration, sleep }
--
+    }
--
+}
--
 /// Future returned by a type-erased transport factory.
 type TransportFuture = Pin<Box<dyn Future<Output = Result<BoxConn, BoxError>> + Send + 'static>>;
--
 /// Type-erased transport construction below HTTP protocol establishment.
 ///
 /// A pool retains one factory that applies partition interface placement,
 /// connector readiness, and the operation's connection timeout. The concrete
 /// connector remains responsible for DNS, proxy, TLS, socket construction,
 /// and transport metadata. Protocol establishment consumes the resulting
 /// [`BoxConn`].
 pub(super) trait TransportFactory: Send + Sync + 'static {
     /// Returns whether this factory can guarantee HTTP/1 for a required attempt.
     fn guarantees_http1(&self) -> bool;
--
     /// Creates one partition-bound transport for `uri`.
     ///
     /// Connector readiness completes before `timeout` starts.
     fn connect(
         &self,
         partition: &PartitionState,
         uri: Uri,
         timeout: Option<TransportTimeout>,
         alpn_protocols: AlpnProtocols,
     ) -> TransportFuture;
--
+}
--
 /// Generic connector factory erased at pool construction.
 struct ServiceTransportFactory<F> {
     /// Builds a connector with the selected network-interface binding.
     connector_for_interface: F,
     /// Whether every H1-required connection is guaranteed to negotiate H1.
     guarantees_http1: bool,
--
+}
--
 impl<F, C, IO> TransportFactory for ServiceTransportFactory<F>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     fn guarantees_http1(&self) -> bool {
         self.guarantees_http1
--
+    }
--
     fn connect(
         &self,
         partition: &PartitionState,
         uri: Uri,
         timeout: Option<TransportTimeout>,
         _alpn_protocols: AlpnProtocols,
     ) -> TransportFuture {
         let interface = partition.interface().map(|interface| interface.as_ref());
         let mut connector = (self.connector_for_interface)(interface);
         Box::pin(async move {
             // Readiness follows the existing connector contract and is not timed.
             poll_fn(|cx| connector.poll_ready(cx))
                 .await
                 .map_err(Into::into)?;
             let connect = connector.call(uri);
             let io = timeout::maybe_timeout_future(
                 connect,
                 timeout.as_ref().map(|timeout| timeout.duration),
                 timeout.as_ref().map(|timeout| &timeout.sleep),
                 TimeoutKind::Connect,
--
+            )
             .await?;
             Ok(Box::new(io) as BoxConn)
         })
--
+    }
--
+}
--
 #[cfg(any(
     all(feature = "test-util", aws_sdk_unstable),
     all(test, feature = "rt-tokio")
 ))]
 /// Erases a clonable connector for pool storage.
 pub(super) fn transport_factory<C, IO>(connector: C) -> StdArc<dyn TransportFactory>
 where
     C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     service_transport_factory_for_interface(move |_| connector.clone(), false)
--
+}
--
 /// Erases a connector constructor that applies an optional interface binding.
 ///
 /// The default HTTP connector consumes this value. Custom connectors use
 /// `transport_factory` and ignore pool interface placement.
 pub(super) fn transport_factory_for_interface<F, C, IO>(
     connector_for_interface: F,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     service_transport_factory_for_interface(connector_for_interface, true)
--
+}
--
 /// Erases a service connector and its HTTP/1 negotiation guarantee.
 fn service_transport_factory_for_interface<F, C, IO>(
     connector_for_interface: F,
     guarantees_http1: bool,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     StdArc::new(ServiceTransportFactory {
         connector_for_interface,
         guarantees_http1,
     })
--
+}
--
 /// Erases and caches connectors whose construction carries TLS or other setup.
 ///
 /// Interface placement is immutable for a partition. The first request for
 /// each `(interface, ALPN offer)` constructs its connector while holding the
 /// cache lock; later requests clone that retained connector. Provider
 /// configuration and certificate loading therefore occur at most once for
 /// each placement and offer rather than once per connection.
 #[cfg(any(feature = "__rustls", feature = "s2n-tls"))]
 pub(super) fn cached_transport_factory_for_interface<F, C, IO>(
     connector_for_interface: F,
     guarantees_http1: bool,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>, AlpnProtocols) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     struct Cached<F, C> {
         factory: F,
         guarantees_http1: bool,
         connectors: crate::sync::Mutex<HashMap<(Option<String>, AlpnProtocols), C>>,
--
+    }
--
     impl<F, C, IO> TransportFactory for Cached<F, C>
     where
         F: Fn(Option<&str>, AlpnProtocols) -> C + Send + Sync + 'static,
         C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
         C::Error: Into<BoxError>,
         C::Future: Send + 'static,
         IO: AsyncConn,
--
+    {
         fn guarantees_http1(&self) -> bool {
             self.guarantees_http1
--
+        }
--
         fn connect(
             &self,
             partition: &PartitionState,
             uri: Uri,
             timeout: Option<TransportTimeout>,
             alpn_protocols: AlpnProtocols,
         ) -> TransportFuture {
             let interface = partition.interface().map(|value| value.to_string());
             let mut connector = self
                 .connectors
                 .lock()
                 .entry((interface.clone(), alpn_protocols))
                 .or_insert_with(|| (self.factory)(interface.as_deref(), alpn_protocols))
                 .clone();
             Box::pin(async move {
                 poll_fn(|cx| connector.poll_ready(cx))
                     .await
                     .map_err(Into::into)?;
                 let connect = connector.call(uri);
                 let io = timeout::maybe_timeout_future(
                     connect,
                     timeout.as_ref().map(|timeout| timeout.duration),
                     timeout.as_ref().map(|timeout| &timeout.sleep),
                     TimeoutKind::Connect,
--
+                )
                 .await?;
                 Ok(Box::new(io) as BoxConn)
             })
--
+        }
--
+    }
--
     StdArc::new(Cached {
         factory: connector_for_interface,
         guarantees_http1,
         connectors: crate::sync::Mutex::new(HashMap::new()),
     })
--
+}
-37
 /// Result of one owner-runtime establishment task.
 pub(super) enum EstablishmentOutcome {
     /// The launching waiter receives this terminal result.
     Complete(AcquisitionResult),
     /// Waiter completion transferred to an H2 flight or generation.
     Transferred,
     Complete(AcquisitionOutcome),
     /// An H2 flight or generation now owns the launching waiter's completion.
     WaiterCompletionTransferred,
-44
+}
-45
 /// Connects one transport and dispatches protocol establishment after ALPN.
 pub(super) async fn establish(
     context: AcquisitionContext,
     waiter: WaiterId,
     permit: EstablishmentPermit,
     requirement: ProtocolRequirement,
 ) -> EstablishmentOutcome {
     let io = match context
         .pool
         .transport
         .connect(
     let connect = TransportConnectContext::new(
         &context.partition,
         context.absolute_uri.clone(),
         context.connect_timeout.clone(),
             alpn_protocols(requirement),
--
+        )
         .await
--
+    {
         requirement,
     );
     let io = match context.pool.transport.connect(connect).await {
         Ok(io) => io,
         Err(error) => {
             tracing::debug!(
                 request_partition = ?context.partition.id(),
                 connection_partition = ?context.cell.id().partition(),
                 origin_scheme = %context.cell.id().origin().scheme(),
                 origin_host = context.cell.id().origin().host(),
                 origin_port = ?context.cell.id().origin().port(),
                 error = ?error,
                 "transport establishment failed"
             );
             return EstablishmentOutcome::Complete(AcquisitionResult::Failed(
                 super::super::downcast_error(error),
             ));
             return EstablishmentOutcome::Complete(AcquisitionOutcome::Failed(downcast_error(
                 error,
             )));
-74
+        }
     };
     let connected = io.connected();
     let negotiated_h2 = connected.is_negotiated_h2();
     tracing::debug!(
         request_partition = ?context.partition.id(),
         connection_partition = ?context.cell.id().partition(),
         origin_scheme = %context.cell.id().origin().scheme(),
         origin_host = context.cell.id().origin().host(),
         origin_port = ?context.cell.id().origin().port(),
         negotiated_protocol = if negotiated_h2 { "HTTP/2" } else { "HTTP/1.1" },
         "transport protocol negotiated"
     );
     if negotiated_h2 && !requirement.accepts_h2() {
         drop(io);
         drop(permit);
         return EstablishmentOutcome::Complete(AcquisitionResult::Failed(ConnectorError::user(
             NegotiatedProtocolMismatch { requirement }.into(),
         )));
         return EstablishmentOutcome::Complete(AcquisitionOutcome::Failed(
             negotiated_protocol_mismatch(requirement),
         ));
-93
+    }
-94
     if negotiated_h2 {
         h2::establish_h2(context, waiter, permit, io, connected).await
     } else {
         EstablishmentOutcome::Complete(
             h1::establish_h1(context, permit, io, connected)
                 .await
                 .map(AcquisitionResult::H1)
                 .unwrap_or_else(AcquisitionResult::Failed),
                 .map(AcquisitionOutcome::H1)
                 .unwrap_or_else(AcquisitionOutcome::Failed),
-103
+        )
-104
+    }
-105
+}
--
 /// An established transport selected a protocol incompatible with the request.
 #[derive(Debug)]
 struct NegotiatedProtocolMismatch {
     requirement: ProtocolRequirement,
-110
+}
-111
 impl fmt::Display for NegotiatedProtocolMismatch {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
             "transport negotiated HTTP/2, which does not satisfy {:?} request semantics",
             self.requirement
-118
+        )
-119
+    }
-120
+}
-121
 impl Error for NegotiatedProtocolMismatch {}
-123
 fn negotiated_protocol_mismatch(requirement: ProtocolRequirement) -> ConnectorError {
     ConnectorError::other(NegotiatedProtocolMismatch { requirement }.into(), None)
-+
+}
-+
 /// Mints one non-wrapping physical-connection identity.
 fn next_connection_id(pool: &PoolInner) -> Result<ConnectionId, ConnectionIdExhausted> {
     let value = pool
         .next_connection_id
         .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
             current.checked_add(1)
         })
         .map_err(|_| ConnectionIdExhausted)?;
     Ok(ConnectionId::new(value))
-137
+}
-138
 /// The pool exhausted its monotonic physical-connection identity space.
 #[derive(Debug)]
 struct ConnectionIdExhausted;
-142
 impl fmt::Display for ConnectionIdExhausted {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str("connection identifier space exhausted")
-146
+    }
-147
+}
-148
 impl Error for ConnectionIdExhausted {}
-150
 /// Static ALPN protocol offer used for one transport connection.
 pub(super) type AlpnProtocols = &'static [&'static [u8]];
--
 /// Default offer for requests that may use HTTP/2.
 pub(super) const HTTP_ALPN_PROTOCOLS: AlpnProtocols = &[b"h2", b"http/1.1"];
 #[cfg(test)]
 mod tests {
     use super::*;
-154
 /// Narrowed offer for requests that require HTTP/1 wire semantics.
 pub(super) const HTTP1_ALPN_PROTOCOLS: AlpnProtocols = &[b"http/1.1"];
     #[test]
     fn negotiated_protocol_mismatch_is_not_a_user_error() {
         let error = negotiated_protocol_mismatch(ProtocolRequirement::H1Required);
-158
 fn alpn_protocols(requirement: ProtocolRequirement) -> AlpnProtocols {
     match requirement {
         ProtocolRequirement::H1Required => HTTP1_ALPN_PROTOCOLS,
         ProtocolRequirement::H1Compatible | ProtocolRequirement::H2Required => HTTP_ALPN_PROTOCOLS,
         assert!(error.is_other());
         assert!(!error.is_user());
-161
+    }
-162
+}

Pages: 1 2 3 4 5 6 7 8 9 10 11