AWS SDK

rev. 39e8f0a4a4cd7fc94cae1e2f796e9c21897d9cb9

Files changed:

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

@@ -25,25 +167,188 @@

 //! Idle(sender) ------------ begin_close ----------------------> Closing
 //! Selected / ReservedForPeer -- begin_close / close_owned ---> Closing
 //! Closing -------- remove_closed -----------------------------> removed
 //! ```
 //!
 //! Outside the lock, ownership moves through values whose drop behavior is
 //! specific to the protocol phase:
 //!
 //! ```text
 //! H1Selection
 //!   |-- dropped ------------------------------> return to connection-owning cell
 //!   |-- retire_connection --------------------------------> logical close
 //!   `-- request accepted --> H1Exchange
 //!                              |-- incomplete or dropped --> logical close
 //!                              `-- offer_for_reuse --------> return to connection-owning cell
 //!
 //! cross-cell reuse --> ProvisionalH1
 //!                        |-- borrow ----------------------> H1Selection
 //!                        |-- reclaim ---------------------> logical close
 //!                        `-- rejected or dropped ---------> return to connection-owning cell

 //! ```
 //!
 //! [`H1Selection`], [`H1Exchange`], and [`ProvisionalH1`] are the
 //! sender-owning values outside the cell lock. Their drop behavior returns or
 //! retires the sender so a cancellation cannot leave a record with no terminal
 //! owner. They retain the owning cell weakly because a selected sender may
 //! temporarily be the ready result in that same cell; cell teardown makes the
 //! fallback close the connection directly.
-53
 use super::super::admission::{
     AdmissionAction, H1Candidate, H1MatchId, H1SupplyStatus, OriginAdmission,
     AdmissionAction, H1Candidate, H1MatchId, H1SupplyStatus, OriginAdmission, PreparedH1IdleProbe,
     PreparedH1Reservation, SupplyRevision,
 };
 use super::super::connection::{CloseReason, ConnectionState};
 use super::{AcquisitionOutcome, OriginCell};
 use crate::sync::{Arc, Weak};
 use aws_smithy_runtime_api::client::connection::ConnectionId;
 use std::collections::{HashMap, VecDeque};
 use std::fmt;
 use std::task::{Context, Poll};
 use std::time::SystemTime;
-66
 use aws_smithy_types::body::SdkBody;
-68
 /// Exclusive request handle for one Hyper HTTP/1 client connection.
 ///
 /// Hyper creates one `SendRequest<SdkBody>` handle when the client handshake
 /// completes. Pool code calls this handle the HTTP/1 sender. Moving it between
 /// residence guards transfers authority to send the next request; it does not
 /// move the socket or protocol driver. The test-only variant exercises those
 /// ownership transitions without running Hyper.
 pub(in crate::client::pool) enum H1Sender {
     /// Hyper's exclusive HTTP/1 request sender.
     Hyper(hyper::client::conn::http1::SendRequest<SdkBody>),
     /// Synthetic sender identity used only by ownership tests.
     #[cfg(test)]
     Test(u64),
     Test { id: u64, ready: bool },
-82
+}
-83
 impl H1Sender {
     /// Wraps a sender returned by a successful Hyper HTTP/1 handshake.
     pub(in crate::client::pool) fn from_hyper(
         sender: hyper::client::conn::http1::SendRequest<SdkBody>,
     ) -> Self {
         Self::Hyper(sender)
-90
+    }
-91
     /// Returns the Hyper sender for readiness and dispatch.
     ///
     /// # Panics
     ///
     /// Panics when a test-only sender reaches the real dispatch path.
     pub(in crate::client::pool) fn hyper_mut(
         &mut self,
     ) -> &mut hyper::client::conn::http1::SendRequest<SdkBody> {
         match self {
             Self::Hyper(sender) => sender,
             #[cfg(test)]
             Self::Test(_) => panic!("test HTTP/1 sender reached Hyper dispatch"),
             Self::Test { .. } => panic!("test HTTP/1 sender reached Hyper dispatch"),
-104
+        }
-105
+    }
-106
     /// Returns whether Hyper already permits another request.
     fn is_ready(&self) -> bool {
         match self {
             Self::Hyper(sender) => sender.is_ready(),
             #[cfg(test)]
             Self::Test(_) => panic!("test HTTP/1 sender reached Hyper readiness"),
             Self::Test { ready, .. } => *ready,
-+
+        }
-+
+    }
-+
     /// Polls Hyper for proof that another request may be sent.
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), hyper::Error>> {
         match self {
             Self::Hyper(sender) => sender.poll_ready(cx),
             #[cfg(test)]
             Self::Test { ready: true, .. } => Poll::Ready(Ok(())),
             #[cfg(test)]
             Self::Test { ready: false, .. } => Poll::Pending,
-124
+        }
-125
+    }
-126
     /// Creates a synthetic sender for state-machine tests.
     #[cfg(test)]
     pub(in crate::client::pool) fn test(id: u64) -> Self {
         Self::Test(id)
         Self::Test { id, ready: true }
-+
+    }
-+
     /// Creates a synthetic sender whose readiness remains pending.
     #[cfg(test)]
     pub(in crate::client::pool) fn pending_test(id: u64) -> Self {
         Self::Test { id, ready: false }
-137
+    }
-138
     /// Returns the synthetic sender identity.
     #[cfg(test)]
     pub(super) fn test_id(&self) -> u64 {
         match self {
             Self::Test(id) => *id,
             Self::Test { id, .. } => *id,
             Self::Hyper(_) => panic!("Hyper sender used in a synthetic ownership test"),
-145
+        }
-146
+    }
-147
+}
-148
 impl fmt::Debug for H1Sender {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::Hyper(_) => f.write_str("H1Sender::Hyper"),
             #[cfg(test)]
             Self::Test(id) => f.debug_tuple("H1Sender::Test").field(id).finish(),
             Self::Test { id, ready } => f
                 .debug_struct("H1Sender::Test")
                 .field("id", id)
                 .field("ready", ready)
                 .finish(),
-159
+        }
-160
+    }
-161
+}
-162
 /// Local reservation for one peer match and its fairness debt.
 ///
 /// Reservation residence and fairness debt are separate because a completed
 /// transfer releases the reservation immediately, while the debt remains until
 /// later local service or the disappearance of compatible local demand.
 #[derive(Debug, Default)]

↧ Expand context

 struct H1Reservation {
     /// State of the current peer reservation.
     state: H1ReservationState,
     /// Whether this cell's next usable HTTP/1 turn must remain local after the
     /// operation that earned it has completed.
     local_turn_owed: bool,
-175
+}
-176
 /// Authoritative state of one cell-local peer reservation.
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
 enum H1ReservationState {
     /// No peer match may intercept a sender return.
     #[default]
     Available,
     /// The next reusable sender return is reserved for this match.
     Installed(H1MatchId),
     /// A provisional sender is outside the cell lock for this match.
     Resolving(H1MatchId),
-187
+}
-188

@@ -610,631 +669,704 @@

↥ Expand context

     pub(super) fn remove_closed(&mut self, id: ConnectionId) {
         let Some(record) = self.records.get(&id) else {
             return;
         };
         if !matches!(record.sender_state, H1SenderResidence::Closing) {
             return;
-637
+        }
         self.records.remove(&id);
         self.assert_consistent();
-640
+    }
-641
     /// Returns idle and externally owned sender counts.
     pub(super) fn connection_counts(&self) -> (usize, usize) {
         self.records
             .values()
             .fold((0, 0), |(idle, active), record| match record.sender_state {
                 H1SenderResidence::Idle { .. } => (idle + 1, active),
                 H1SenderResidence::Selected | H1SenderResidence::ReservedForPeer => {
                     (idle, active + 1)
-650
+                }

                 H1SenderResidence::Closing => (idle, active),
             })
-653
+    }
-654
     /// Returns the installed record and idle counts.
     #[cfg(test)]
     pub(super) fn counts(&self) -> (usize, usize) {
         (self.records.len(), self.idle_order.len())
-659
+    }
-660
     /// Returns the synthetic identities of every idle sender.
     #[cfg(all(test, smithy_http_client_loom))]
     pub(super) fn idle_sender_ids(&self) -> Vec<u64> {
         self.records
             .values()
             .filter_map(|record| match &record.sender_state {
                 H1SenderResidence::Idle { sender, .. } => Some(sender.test_id()),
                 H1SenderResidence::Selected
                 | H1SenderResidence::ReservedForPeer
                 | H1SenderResidence::Closing => None,
             })
             .collect()
-+
+    }
-+
     /// Returns the sole installed connection for focused dispatch tests.
     #[cfg(all(test, feature = "rt-tokio"))]
     pub(super) fn only_connection_for_test(&self) -> Arc<ConnectionState> {
         assert_eq!(
 ,
             self.records.len(),
             "expected exactly one installed HTTP/1 record"
         );
         self.records
             .values()

↧ Expand context

             .next()
             .expect("HTTP/1 record count changed under the cell lock")
             .connection
             .clone()
-689
+    }
-690
     /// Returns idle records whose configured deadline has elapsed.
     pub(super) fn expired_idle(&self, now: SystemTime) -> Vec<ConnectionId> {
         self.idle_order
             .iter()
             .copied()
             .filter(|id| {
                 matches!(
                     self.records.get(id).map(|record| &record.sender_state),
                     Some(H1SenderResidence::Idle {
                         deadline: Some(deadline),
                         ..
                     }) if *deadline <= now
-703
+                )
             })

@@ -927,962 +1005,1047 @@

↥ Expand context

-962
+            )
             .field("connection_id", &self.owner.as_ref().map(OwnedH1Sender::id))
             .finish()
-965
+    }
-966
+}
-967
 impl Drop for H1Selection {
     fn drop(&mut self) {
         if let Some(owner) = self.owner.take() {
             return_to_connection_cell(&self.connection_cell, owner);
-972
+        }
-973
+    }
-974
+}
-975
 /// Exclusive sender for a request Hyper accepted on this connection.
 ///
 /// The installed record remains `Selected` while response-body processing owns
 /// this value. After a complete response and successful Hyper readiness,
 /// [`H1Exchange::offer_for_reuse`] runs owning-cell return arbitration.
 /// Explicit retirement handles protocol failure or upgrade. Dropping the

 /// exchange means a reusable message boundary was not proven and closes the
 /// connection as an incomplete HTTP/1 exchange.
 pub(in crate::client::pool) struct H1Exchange {
     /// Non-retaining reference to the cell that owns the selected record.
     connection_cell: Weak<OriginCell>,
     /// Sender held until Hyper proves it may return.
     owner: Option<OwnedH1Sender>,
-989
+}
-990
 impl H1Exchange {
     /// Returns the installed connection that owns this exchange.
     pub(in crate::client::pool) fn connection(&self) -> &Arc<ConnectionState> {
         self.owner
             .as_ref()
             .expect("HTTP/1 exchange consumed more than once")
             .connection()
-+
+    }
-+
     /// Returns whether Hyper already permits another request.
     pub(in crate::client::pool) fn is_ready(&self) -> bool {
         self.owner
             .as_ref()
             .expect("HTTP/1 exchange consumed more than once")
             .sender
             .is_ready()
-1007
+    }
-1008
     /// Polls Hyper for proof that the sender can accept another request.
     pub(in crate::client::pool) fn poll_ready(
         &mut self,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), hyper::Error>> {
         self.owner
             .as_mut()
             .expect("HTTP/1 exchange consumed more than once")
             .sender_mut()
             .hyper_mut()
             .poll_ready(cx)
-1019
+    }
-1020
     /// Offers a proven-ready sender to owning-cell reuse policy.
     pub(in crate::client::pool) fn offer_for_reuse(mut self) {
         let mut owner = self
             .owner
             .take()
             .expect("HTTP/1 exchange consumed more than once");
         let connection = owner.connection().clone();

↧ Expand context

         owner.mark_reused();
         return_to_connection_cell(&self.connection_cell, owner);
         connection.complete_h1_exchange(CloseReason::ProtocolClosed);
-1031
+    }
-1032
     /// Retires the sender instead of returning it to the connection-owning cell.
     pub(in crate::client::pool) fn retire_connection(mut self, reason: CloseReason) {
         if let Some(owner) = self.owner.take() {
             let connection = owner.connection().clone();
             retire_at_connection_cell(&self.connection_cell, owner, reason);
             connection.complete_h1_exchange(reason);
-1039
+        }
-1040
+    }
-1041
+}
-1042
 impl fmt::Debug for H1Exchange {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("H1Exchange")
             .field(
                 "connection-owning cell",

@@ -1248,1290 +1307,1383 @@

↥ Expand context

-1290
+    }
-1291
     /// Commits one admission-selected supplier reservation at this cell.
     pub(in crate::client::pool) fn commit_h1_reservation(
         cell: &Arc<Self>,
         admission: Arc<OriginAdmission>,
         prepared: PreparedH1Reservation,
     ) -> Option<AdmissionAction> {
         let decision = {
             let mut state = cell.state.lock();
             state.commit_h1_reservation(prepared.match_id)
         };
-1302
         let decision = decision.map_candidate(|owner| {
             let provisional = ProvisionalH1::new(cell, owner);
             H1Candidate::new(
                 admission.clone(),
                 prepared.match_id,
                 cell.id.partition(),
                 provisional,

-1310
+            )
         });
         OriginAdmission::settle_h1_reservation(
             &admission,
             prepared.match_id,
             cell.id.partition(),
             decision,
-1317
+        )
-1318
+    }
-1319
     /// Attempts to extract an idle sender without intercepting a future return.
     pub(in crate::client::pool) fn try_take_idle_h1(
         cell: &Arc<Self>,
         admission: Arc<OriginAdmission>,
         prepared: PreparedH1IdleProbe,
     ) -> Option<AdmissionAction> {
         let decision = {
             let mut state = cell.state.lock();
             state.try_take_idle_h1(prepared.match_id)
         };
         let decision = decision.map_candidate(|owner| {
             let provisional = ProvisionalH1::new(cell, owner);
             H1Candidate::new(
                 admission.clone(),
                 prepared.match_id,
                 cell.id.partition(),
                 provisional,
-+
+            )
         });
         OriginAdmission::settle_h1_idle_probe(
             &admission,
             prepared.match_id,
             cell.id.partition(),
             decision,
-+
+        )
-+
+    }
-+
     /// Returns the current admission-facing HTTP/1 supply revision.
     pub(in crate::client::pool) fn current_h1_supply_revision(
         &self,
     ) -> SupplyRevision<H1SupplyStatus> {
         self.state.lock().current_h1_supply_revision()
-+
+    }
-+
     /// Clears an installed or resolving reservation after request cancellation.
     pub(in crate::client::pool) fn cancel_h1_reservation(
         &self,
         match_id: H1MatchId,
     ) -> SupplyRevision<H1SupplyStatus> {
         self.state.lock().cancel_h1_reservation(match_id)
-1360
+    }
-1361
     /// Returns a rejected provisional sender through ordinary connection-owning cell handling.
     pub(in crate::client::pool) fn reject_h1_match(

↧ Expand context

         cell: &Arc<Self>,
         match_id: H1MatchId,
         provisional: ProvisionalH1,
     ) -> SupplyRevision<H1SupplyStatus> {
-1368
+        {
             let mut state = cell.state.lock();
             state.reject_h1_match(match_id);
-1371
+        }
         drop(provisional);
         let mut state = cell.state.lock();
         state.assert_consistent();
         state.current_h1_supply_revision()
-1376
+    }
-1377
     /// Revalidates one match and commits its provisional sender for dispatch.
     pub(in crate::client::pool) fn commit_h1_match(
         cell: &Arc<Self>,
         match_id: H1MatchId,
         provisional: ProvisionalH1,
     ) -> Result<H1Selection, ProvisionalH1> {

@@ -1540,1616 +1599,1681 @@

↥ Expand context

     ///
     /// Returns whether this signal won the connection's logical-close race.
     pub(super) fn close_h1(cell: &Arc<Self>, id: ConnectionId, reason: CloseReason) -> bool {
         let Some((close, supply_update)) = ({
             let mut state = cell.state.lock();
             let close = state.h1.begin_close(id);
             state.assert_consistent();
             close.map(|close| (close, state.take_h1_supply_update()))
         }) else {
             return false;
         };
-1627
         cell.submit_h1_supply_update(supply_update);
         let remove_record = close.sender.is_some();
         let won = close.connection.logical_close(reason);
         drop(close.sender);
         if remove_record {
             let mut state = cell.state.lock();
             state.h1.remove_closed(id);
             state.assert_consistent();

-1636
+        }
         won
-1638
+    }
-1639
     /// Returns H1 record counts for focused ownership tests.
     #[cfg(test)]
     pub(super) fn h1_counts(&self) -> (usize, usize) {
         self.state.lock().h1.counts()
-1644
+    }
-1645
     /// Returns the synthetic identities of every idle HTTP/1 sender.
     #[cfg(all(test, smithy_http_client_loom))]
     pub(super) fn h1_idle_sender_ids(&self) -> Vec<u64> {
         self.state.lock().h1.idle_sender_ids()
-+
+    }
-+
     /// Returns the sole installed HTTP/1 connection for focused dispatch tests.
     #[cfg(all(test, feature = "rt-tokio"))]
     pub(in crate::client::pool) fn only_h1_connection_for_test(&self) -> Arc<ConnectionState> {
         self.state.lock().h1.only_connection_for_test()
-1656
+    }
-1657
+}
-1658
 /// Terminal outcome for one reusable HTTP/1 sender return.
 enum H1ReturnTrace {
     /// The owning cell no longer accepts the sender.

↧ Expand context

     Rejected,
     /// Origin admission reserved the sender for peer demand.
     PeerReservationIntercepted,
     /// No compatible demand exists, so the sender became idle.
     Idle,
     /// A waiter in the owning cell accepted the sender.
     LocalDemand,
-1669
+}
-1670
 /// Emits a committed HTTP/1 return outcome outside the cell lock.
 // Keep field formatting out of a return transition that can synchronously
 // enter admission and sender fallbacks.
 #[inline(never)]
 fn trace_h1_return(cell: &OriginCell, connection_id: ConnectionId, outcome: H1ReturnTrace) {
     match outcome {
         H1ReturnTrace::Rejected => tracing::trace!(
             connection_id = %connection_id,
             connection_partition = ?cell.id.partition(),
             origin_scheme = %cell.id.origin().scheme(),
             origin_host = cell.id.origin().host(),

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

@@ -2895,2895 +3050,3103 @@

↥ Expand context

             .expect("idle transition did not resume H1-required demand")
         else {
             panic!("idle HTTP/2 reclaim produced a non-capacity result");
         };
         assert_eq!(
             Some(CloseReason::Reclaimed),
             connection.probe().close_reason
         );
         drop(permit);
         assert_eq!(1, admission.available_capacity_for_test());
-2905
+    }
-2906
     #[test]
     fn pool_shutdown_closes_an_accepting_generation() {
         let cell = cell();
         let (_generation, connection, _physical) = open_test_generation(&cell, 1);
-2911
         OriginCell::close_all(&cell, CloseReason::PoolDropped);
-2913
         assert_eq!(

             Some(CloseReason::PoolDropped),
             connection.probe().close_reason
         );
         assert_eq!(None, cell.accepting_h2_generation());
-2919
+    }
-2920
+}
-2921
 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use crate::client::pool::admission::ProtocolRequirement;
     use crate::client::pool::connection::{CloseReason, ConnectionInfo};
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::EligibilityGroup;
     use aws_smithy_runtime_api::client::connection::ConnectionId;
     use http_1x::uri::Scheme;
-2931
     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,
         ))
-2940
+    }
-2941
     fn open_test_generation(cell: &Arc<OriginCell>) -> (H2GenerationId, Arc<ConnectionState>) {
         let (connection, _physical) = ConnectionState::unbounded(ConnectionInfo::for_test(
             ConnectionId::new(1),
             PartitionId::from_index(1),
         ));
         let generation = OriginCell::install_h2_for_test(cell, connection.clone(), 1, None);
         (generation, connection)
-2949
+    }
-2950
     fn launching_waiter(cell: &Arc<OriginCell>) -> WaiterId {
         let waiter = OriginCell::register_waiter(cell, ProtocolRequirement::H2Required);
         let AcquisitionStep::StartEstablishment(permit) = cell
             .take_ready_event(waiter)
             .expect("new HTTP/2 waiter did not receive establishment authority")
         else {
             panic!("new HTTP/2 waiter completed before establishment");
         };
         assert!(cell.start_establishment(waiter));
         drop(permit);
         waiter
-+
+    }
-+
     /// Races activation of one exact generation with close of that generation.
     ///
     /// Close either rejects activation or retains the draining generation until
     /// the prospective request releases its claim.
     #[test]
     fn activation_linearizes_against_generation_close() {
         loom::model(|| {
             let cell = cell();
             let (generation, connection) = open_test_generation(&cell);
             let activating_cell = cell.clone();
             let activation = loom::thread::spawn(move || {
                 OriginCell::activate_h2(
                     &activating_cell,
                     generation,
                     activating_cell.id().partition(),
-2979
+                )
             });
             let closing_cell = cell.clone();
             let close = loom::thread::spawn(move || {
                 OriginCell::close_h2(&closing_cell, generation, CloseReason::Poisoned)
             });
-2985
             drop(activation.join().unwrap());
             let activation = activation.join().unwrap();
             assert!(close.join().unwrap());
             assert_eq!(Some(CloseReason::Poisoned), connection.probe().close_reason);
             if let Some(activation) = activation {
                 assert_eq!(
                     Some((1, 0)),
                     cell.h2_request_counts(generation),
                     "close did not retain the prospective activation"
                 );
                 drop(activation);
-+
+            }
             assert!(!cell.state.lock().h2.generations.contains_key(&generation));
         });
-2999
+    }
-3000
     /// Completes the upload and response sides of one accepted request concurrently.
     ///
     /// The two guards share one request claim and must release its dispatch once.
     #[test]
     fn concurrent_request_side_completion_releases_one_dispatch() {
         loom::model(|| {
             let cell = cell();
             let (generation, connection) = open_test_generation(&cell);
             let mut activation = OriginCell::activate_h2(&cell, generation, cell.id().partition())
                 .expect("generation did not activate");
             let H2DispatchParts {
                 sender: _sender,
                 upload,
                 response,
             } = activation.take_dispatch_parts();
             let dispatch = ConnectionState::try_commit_dispatch(&connection)
                 .expect("open connection rejected dispatch");
             activation.accept(dispatch);
-3019
             let upload = loom::thread::spawn(move || drop(upload));
             let response = loom::thread::spawn(move || drop(response));
             upload.join().unwrap();
             response.join().unwrap();
-3024
             assert_eq!(0, connection.probe().in_flight);
             let state = cell.state.lock();
             let record = state
                 .h2
                 .generations
                 .get(&generation)
                 .expect("accepting generation disappeared");
             assert_eq!(0, record.active_requests);
         });
-3034
+    }
-3035
     /// Converges two post-ALPN attempts, then races flight completion with cancellation.
     ///
     /// Exactly one attempt owns the flight task. The live participant receives
     /// the installed generation while the cancelled participant is removed from
     /// either the flight or the generation gate.
     #[test]
     fn generation_close_waits_for_both_request_sides() {
     fn concurrent_h2_flight_convergence_installs_one_generation() {
         loom::model(|| {
             let cell = cell();
             let (connection, _physical) = ConnectionState::unbounded(ConnectionInfo::for_test(
             let first = launching_waiter(&cell);
             let second = launching_waiter(&cell);
-+
             let first_cell = cell.clone();
             let first_convergence =
                 loom::thread::spawn(move || first_cell.converge_h2_flight(first));
             let second_cell = cell.clone();
             let second_convergence =
                 loom::thread::spawn(move || second_cell.converge_h2_flight(second));
             let first_decision = first_convergence.join().unwrap();
             let second_decision = second_convergence.join().unwrap();
             let flight = match (first_decision, second_decision) {
                 (H2FlightDecision::RunFlight(flight), H2FlightDecision::JoinedFlight)
                 | (H2FlightDecision::JoinedFlight, H2FlightDecision::RunFlight(flight)) => flight,
                 decisions => panic!("convergence did not select one flight owner: {decisions:?}"),
             };
-+
             let (connection, physical) = ConnectionState::unbounded(ConnectionInfo::for_test(
                 ConnectionId::new(1),
                 PartitionId::from_index(1),
             ));
             let generation = OriginCell::install_h2_for_test(&cell, connection.clone(), 1, None);
             let mut activation = OriginCell::activate_h2(&cell, generation, cell.id().partition())
                 .expect("generation did not activate");
             let H2DispatchParts {
                 sender: _sender,
                 upload,
                 response,
             } = activation.take_dispatch_parts();
             let dispatch = ConnectionState::try_commit_dispatch(&connection)
                 .expect("open connection rejected dispatch");
             activation.accept(dispatch);
--
             let first_side = loom::thread::spawn(move || drop(upload));
             let closing_cell = cell.clone();
             let close = loom::thread::spawn(move || {
                 OriginCell::close_h2(&closing_cell, generation, CloseReason::ProtocolClosed)
             let completing_cell = cell.clone();
             let completing_connection = connection.clone();
             let completing = loom::thread::spawn(move || {
                 OriginCell::complete_h2_flight(
                     &completing_cell,
                     flight,
                     completing_connection,
                     H2Sender::test(1),
                     None,
-+
+                )
             });
             first_side.join().unwrap();
             assert!(close.join().unwrap());
             let cancelling_cell = cell.clone();
             let cancelling =
                 loom::thread::spawn(move || OriginCell::cancel_waiter(&cancelling_cell, second));
-+
             let generation = completing
                 .join()
                 .unwrap()
                 .expect("the sole HTTP/2 flight did not install");
             assert!(cancelling.join().unwrap());
             let AcquisitionStep::Resolved(AcquisitionOutcome::H2(activation)) = cell
                 .take_ready_event(first)
                 .expect("live flight participant did not receive the generation")
             else {
                 panic!("live flight participant received a non-HTTP/2 result");
             };
             drop(activation);
-3093
             assert_eq!(1, connection.probe().in_flight);
             assert_eq!(
                 Some((0, 1)),
                 cell.h2_request_counts(generation),
                 "draining generation released before its second request side"
             );
             drop(response);
             assert_eq!(0, connection.probe().in_flight);
             assert_eq!(
                 None,
                 cell.h2_request_counts(generation),
                 "finished draining generation remained installed"
             );
             assert_eq!(0, cell.retained_waiters_for_test());
             assert!(OriginCell::close_h2(
                 &cell,
                 generation,
                 CloseReason::PoolDropped,
             ));
             drop(physical);
         });
-3102
+    }
-3103
+}

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

@@ -1,1 +46,48 @@

 /*
  * 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::client::{downcast_error, ConnectorCallTimer};
 use crate::sync::Arc;
 use aws_smithy_async::rt::sleep::{default_async_sleep, SharedAsyncSleep};
 use aws_smithy_async::time::SharedTimeSource;
 use aws_smithy_runtime_api::box_error::BoxError;
 use aws_smithy_runtime_api::client::connector_metadata::ConnectorMetadata;
 use aws_smithy_runtime_api::client::http::telemetry::CaptureHttpAttemptTelemetry;
 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, RuntimeComponentsBuilder,
 };
 use aws_smithy_types::config_bag::ConfigBag;
 use std::borrow::Cow;

↧ Expand context

 use std::error::Error;
 use std::fmt;
 use std::time::Duration;
-32
 /// 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>,
-45
+}
-46
 impl Client {
     /// Creates a client for a pool built without explicit partitions.

@@ -79,81 +145,149 @@

↥ Expand context

-81
 impl fmt::Debug for Client {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Client")
             .field("partition", &self.partition.id())
             .field("pool", &self.pool)
             .finish_non_exhaustive()
-88
+    }
-89
+}
-90
 impl HttpClient for Client {
     fn validate_base_client_config(
         &self,
         _: &RuntimeComponentsBuilder,
         _: &ConfigBag,
     ) -> Result<(), BoxError> {
         self.pool
             .inner
             .transport
             .initialize_for_partition(&self.partition)

-101
+    }
-102
     fn http_connector(
         &self,
         settings: &HttpConnectorSettings,
         components: &RuntimeComponents,
     ) -> SharedHttpConnector {
         let connect_timeout = settings.connect_timeout();
         let read_timeout = settings.read_timeout();
         let sleep = components.sleep_impl().or_else(default_async_sleep);
         let time_source = components.time_source().unwrap_or_default();
-112
         SharedHttpConnector::new(PoolConnector {
             pool: self.pool.clone(),
             partition: self.partition.clone(),
             connect_timeout,
             read_timeout,
             sleep,
             time_source,
         })
-121
+    }
-122
     fn connector_metadata(&self) -> Option<ConnectorMetadata> {
         Some(ConnectorMetadata::new("hyper", Some(Cow::Borrowed("1.x"))))
-125
+    }
-126
+}
-127
 /// Error returned when a [`Client`] cannot be resolved from a pool.
 #[derive(Clone, Debug, Eq, PartialEq)]

↧ Expand context

 pub struct ClientBuildError {
     /// Construction failure represented without exposing the internal enum.
     kind: ClientBuildErrorKind,
-133
+}
-134
 impl ClientBuildError {
     /// Creates an unresolved-partition error.
     fn invalid_partition(partition: PartitionId) -> Self {
         Self {
             kind: ClientBuildErrorKind::InvalidPartition(partition),
-140
+        }
-141
+    }
-142
     /// Returns the unresolved partition identity, when applicable.
     pub fn partition(&self) -> Option<PartitionId> {
         match &self.kind {
             ClientBuildErrorKind::InvalidPartition(partition) => Some(*partition),
-147
+        }
-148
+    }
-149
+}

@@ -160,164 +284,318 @@

↥ Expand context

 impl Error for ClientBuildError {}
-165
 /// Internal categories represented by [`ClientBuildError`].
 #[derive(Clone, Debug, Eq, PartialEq)]
 enum ClientBuildErrorKind {
     /// The pool retained no partition with this identity.
     InvalidPartition(PartitionId),
-171
+}
-172
 /// Per-operation Smithy adapter over one resolved pool partition.
 ///
 /// [`HttpClient::http_connector`] combines the retained pool and partition
 /// with that operation's timeout settings. Each call validates timeout timer
 /// availability, converts the Smithy request, and routes it through the shared
 /// pool. The connect timeout covers a newly started transport operation; the
 /// read timeout covers pool acquisition and dispatch through response headers.
 /// The adapter neither creates another pool nor resolves the partition again.
 struct PoolConnector {
     /// Shared pool used for acquisition and dispatch.
     pool: ConnectionPool,

     /// Partition selected when the client was constructed.
     partition: Arc<PartitionState>,
     /// Maximum duration of the transport connection operation.
     ///
     /// Connector readiness completes before this timeout starts.
     connect_timeout: Option<Duration>,
     /// Maximum duration through response headers.
     read_timeout: Option<Duration>,
     /// Runtime timer used by operation timeouts.
     sleep: Option<SharedAsyncSleep>,
     /// Operation runtime clock used for request-attempt telemetry.
     time_source: SharedTimeSource,
-196
+}
-197
 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()
-205
+    }
-206
+}
-207
 /// Converts and dispatches one request through the partitioned pool.
 async fn send_pool_request(
     pool: &ConnectionPool,
     partition: Arc<PartitionState>,
     request: HttpRequest,
     connect_timeout: Option<Duration>,
     read_timeout: Option<Duration>,
     sleep: Option<SharedAsyncSleep>,
     attempt_telemetry: Option<super::dispatch::AttemptTelemetryInput>,
 ) -> Result<HttpResponse, ConnectorError> {
     if (connect_timeout.is_some() || read_timeout.is_some()) && sleep.is_none() {
         return Err(ConnectorError::user(MissingAsyncSleep.into()));
-+
+    }
-+
     let request = request
         .try_into_http1x()
         .map_err(|error| ConnectorError::user(error.into()))?;
     let options = super::dispatch::RequestOptions::new(
         connect_timeout
             .zip(sleep.clone())
             .map(|(duration, sleep)| super::establish::TransportTimeout::new(duration, sleep)),
         attempt_telemetry,
     );
     let send = pool.send_request(partition, request, options);
     let response =
         timeout::maybe_timeout_future(send, read_timeout, sleep.as_ref(), TimeoutKind::Read)
             .await
             .map_err(downcast_error)?;
     HttpResponse::try_from(response).map_err(|error| ConnectorError::other(error.into(), None))
-+
+}
-+
 impl HttpConnector for PoolConnector {
     fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
         let attempt_capture = request.extension::<CaptureHttpAttemptTelemetry>().cloned();
         let connector_call_timer = ConnectorCallTimer::start(&request, &self.time_source);
         let attempt_telemetry = attempt_capture.as_ref().map(|capture| {
             super::dispatch::AttemptTelemetryInput::new(capture.clone(), self.time_source.clone())
         });
         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()));
--
+            }
--
             let request = request.try_into_http1x().map_err(|error| {
                 aws_smithy_runtime_api::client::result::ConnectorError::user(error.into())
             })?;
             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,
             let result = send_pool_request(
                 &pool,
                 partition,
                 request,
                 connect_timeout,
                 read_timeout,
                 sleep.as_ref(),
                 TimeoutKind::Read,
                 sleep,
                 attempt_telemetry,
-260
+            )
             .await
             .map_err(downcast_error)?;
             HttpResponse::try_from(response).map_err(|error| {
                 aws_smithy_runtime_api::client::result::ConnectorError::other(error.into(), None)
             })
             .await;
             if let Some(timer) = connector_call_timer {
                 timer.finish();
-+
+            }
             result
         })
-267
+    }
-268
+}
-269
 /// Operation timeouts were configured without a runtime timer.
 #[derive(Debug)]
 struct MissingAsyncSleep;
-273
 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")
-277
+    }
-278
+}
-279
 impl Error for MissingAsyncSleep {}
-281
 #[cfg(all(test, feature = "rt-tokio", not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
-285
     #[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();
-293
         for (connect_timeout, read_timeout) in [
             (Some(Duration::from_secs(1)), None),
             (None, Some(Duration::from_secs(1))),
         ] {
             let connector = PoolConnector {
                 pool: client.pool.clone(),
                 partition: client.partition.clone(),
                 connect_timeout,
                 read_timeout,
                 sleep: None,
                 time_source: SharedTimeSource::default(),
             };
-306
             let error = connector
                 .call(HttpRequest::get("http://example.com/").unwrap())
                 .await
                 .expect_err("a timeout without sleep unexpectedly started a request");
             assert!(error.is_user(), "unexpected connector error: {error:?}");
             assert_eq!(
                 "an async sleep implementation is required for HTTP connect or read timeouts",
                 std::error::Error::source(&error).unwrap().to_string()

↧ Expand context

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

@@ -1,1 +62,65 @@

↥ Expand context

 /*
  * 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 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 before physical completion except while an HTTP/1
 //! exchange may still transfer upgraded I/O.
-15
 use super::admission::CapacityLease;
 use super::events::{LogicalCloseCause, SharedConnectionEventListener};
 use super::origin::OriginKey;
 use super::partition::PartitionId;
 use super::partition::{DriverSpawner, PartitionId};
 use super::stats::CellConnectionStats;
 use crate::client::connect::{ConnectPath, ConnectPathInner};
 use crate::sync::{Arc, Mutex};
 pub use aws_smithy_runtime_api::client::connection::ConnectionId;
 use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
 use aws_smithy_runtime_api::client::connection::{
     ConnectionEstablishmentMetadata, ConnectionMetadata,
 };
 use http_1x::Extensions;
 use hyper::rt::{Read, ReadBufCursor, Write};
 use hyper_util::client::legacy::connect::{Connected, Connection, HttpInfo};
 use pin_project_lite::pin_project;
 use std::fmt;
 use std::io::{self, IoSlice};
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::sync::{Arc as StdArc, OnceLock};
 use std::task::{Context, Poll};
-37
 /// Final protocol and ownership classification for a closed connection.
 ///
 /// [`LogicalCloseCause`] reports the first stable reason that pool dispatch
 /// ended. This more specific classification is reported at physical close
 /// because HTTP/1 upgrade ownership may be resolved after logical close.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[non_exhaustive]
 pub enum CloseReason {

↧ Expand context

     /// 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.
     Upgraded,
     /// The connection closed to move bounded capacity to another cell.
     Reclaimed,
     /// The connection pool was dropped.
     PoolDropped,
     /// The runtime driving the connection shut down.
     OwnerRuntimeShutdown,
-62
+}
-63
 /// Protocol selected for one installed physical connection.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]

@@ -64,67 +252,286 @@

↥ Expand context

 pub enum ConnectionProtocol {
     /// HTTP/1.1 with one exclusive request sender.
     Http1,
     /// HTTP/2 with a multiplexed request sender.
     Http2,
-72
+}
-73
 /// Immutable identity and transport facts for one installed connection.
 ///
 /// Protocol records, request metadata, tracing, and lifecycle events retain the
 /// same value. It remains valid after logical close so the final physical-close
 /// observation names the same connection.
 #[derive(Debug)]
 pub struct ConnectionInfo {
     /// Stable identity assigned by the owning pool.
     id: ConnectionId,
     /// Canonical origin this connection may serve.
     origin: OriginKey,
     /// Partition that retains the transport and protocol driver.
     owner_partition: PartitionId,

     /// Protocol selected by configuration or ALPN.
     protocol: ConnectionProtocol,
     /// Local socket address reported by the connector, when available.
     local_addr: Option<SocketAddr>,
     /// Remote socket address reported by the connector, when available.
     remote_addr: Option<SocketAddr>,
     /// Connector-owned path state, including request-time proxy authorization.
     connect_path: ConnectPathInner,
     /// Connector metadata copied into every response on this connection.
     connected: Connected,
     /// Measurements frozen before this connection becomes visible to dispatch.
     establishment: OnceLock<ConnectionEstablishmentMetadata>,
-99
+}
-100
 impl ConnectionInfo {
     /// Captures immutable facts after transport negotiation selects a protocol.
     pub(super) fn new(
         id: ConnectionId,
         origin: OriginKey,
         owner_partition: PartitionId,
         protocol: ConnectionProtocol,
         connected: Connected,
     ) -> Arc<Self> {
         let mut extras = Extensions::new();
         connected.get_extras(&mut extras);
         let http_info = extras.get::<HttpInfo>();
         let connect_path = ConnectPathInner::from_connected(&connected, &extras);
         Arc::new(Self {
             id,
             origin,
             owner_partition,
             protocol,
             local_addr: http_info.map(HttpInfo::local_addr),
             remote_addr: http_info.map(HttpInfo::remote_addr),
             connect_path,
             connected,
             establishment: OnceLock::new(),
         })
-125
+    }
-126
     /// Returns the pool-assigned physical connection identity.
     pub fn id(&self) -> ConnectionId {
         self.id
-130
+    }
-131
     /// Returns the canonical origin this connection may serve.
     pub fn origin(&self) -> &OriginKey {
         &self.origin
-135
+    }
-136
     /// Returns the partition that owns this connection's I/O and driver.
     pub fn owner_partition(&self) -> PartitionId {
         self.owner_partition
-140
+    }
-141
     /// Returns the established HTTP protocol.
     pub fn protocol(&self) -> ConnectionProtocol {
         self.protocol
-145
+    }
-146
     /// Returns the connector-reported local socket address, when available.
     pub fn local_addr(&self) -> Option<SocketAddr> {
         self.local_addr
-150
+    }
-151
     /// Returns the connector-reported remote socket address, when available.
     pub fn remote_addr(&self) -> Option<SocketAddr> {
         self.remote_addr
-155
+    }
-156
     /// Returns how this connection reaches its origin.
     pub fn connect_path(&self) -> ConnectPath {
         self.connect_path.public()
-160
+    }
-161
     /// Returns whether an HTTP proxy participates in this connection path.
     pub fn is_proxied(&self) -> bool {
         self.connect_path().is_proxied()
-165
+    }
-166
     /// Returns successful establishment measurements, when installation completed.
     pub fn establishment(&self) -> Option<&ConnectionEstablishmentMetadata> {
         self.establishment.get()
-+
+    }
-+
     /// Freezes successful establishment measurements before pool publication.
     pub(super) fn set_establishment(&self, metadata: ConnectionEstablishmentMetadata) {
         assert!(
             self.establishment.set(metadata).is_ok(),
             "connection establishment metadata was set more than once"
         );
-+
+    }
-+
     /// Returns connector-owned request-path state.
     pub(super) fn connect_path_inner(&self) -> &ConnectPathInner {
         &self.connect_path
-183
+    }
-184
     /// Copies connector-provided values into a response extension map.
     pub(super) fn apply_connector_extras(&self, extensions: &mut Extensions) {
         self.connected.get_extras(extensions);
-188
+    }
-189
     /// Builds Smithy metadata with close authority for this H1 record.
     pub(super) fn metadata(&self, close: super::cell::h1::H1CloseHandle) -> ConnectionMetadata {
         let mut builder = ConnectionMetadata::builder()
             .proxied(self.is_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);
             .set_remote_addr(self.remote_addr)
             .set_establishment(self.establishment().cloned());
         builder.build()
-203
+    }
-204
     /// Builds Smithy metadata with close authority for this H2 generation.
     pub(super) fn h2_metadata(&self, close: super::cell::h2::H2CloseHandle) -> ConnectionMetadata {
         let mut builder = ConnectionMetadata::builder()
             .proxied(self.is_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);
             .set_remote_addr(self.remote_addr)
             .set_establishment(self.establishment().cloned());
         builder.build()
-218
+    }
-219
     /// 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,
             ConnectionProtocol::Http1,
             Connected::new(),
-230
+        )
-231
+    }
-232
+}
-233
 /// 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>,
     /// Runtime that owns protocol and follow-up work for this connection.
     owner_spawner: StdArc<dyn DriverSpawner>,
     /// Cell-owned counts for connection lifetimes that outlive protocol records.
     stats: Arc<CellConnectionStats>,
     /// Dispatch, logical-close, and physical-connection completion state.
     lifecycle: Mutex<ConnectionLifecycle>,
-244
+}
-245
 #[cfg(test)]
 #[derive(Debug)]
 struct TestDriverSpawner;
-+
 #[cfg(test)]
 impl DriverSpawner for TestDriverSpawner {
     fn spawn(&self, driver: Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>) {
         drop(driver);
-+
+    }
-+
+}
-+
 /// Connection lifetime state serialized with dispatch commitment and close.
 #[derive(Debug)]
 struct ConnectionLifecycle {
     /// Dispatch eligibility and ownership of bounded capacity.
     logical: LogicalState,
     /// Requests that committed before logical close.
     in_flight: usize,
     /// Whether the client has released its physical connection handle.
     physical_connection_complete: bool,
     /// Protocol drain or upgrade count held until physical completion.

↧ Expand context

     counted_close: Option<CountedClose>,
     /// Progress through installed-connection event delivery.
     events: LifecycleEventProgress,
-270
+}
-271
 /// 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.
         capacity: Option<CapacityLease>,
     },
     /// New dispatch is rejected while accepted work may still drain.
     Closed {
         /// Stable cause recorded by the first logical-close transition.
         cause: LogicalCloseCause,

@@ -305,339 +493,544 @@

↥ Expand context

-339
+        }
-340
+    }
-341
     fn finish(self, stats: &CellConnectionStats) {
         match self {
             Self::H1Draining => stats.h1_drain_finished(),
             Self::H1Upgraded => stats.h1_upgrade_finished(),
             Self::H2Draining => stats.h2_drain_finished(),
-347
+        }
-348
+    }
-349
     fn h1_upgrade(self, stats: &CellConnectionStats) -> Self {
         assert_eq!(Self::H1Draining, self, "non-HTTP/1 drain became an upgrade");
         stats.h1_drain_upgraded();
         Self::H1Upgraded
-354
+    }
-355
+}
-356
 /// Progress through installed-connection event delivery.
 ///

 /// Connection state may close while the `Opened` callback is running. Tracking
 /// callback progress under the lifecycle lock delays close callbacks until
 /// `Opened` returns and preserves `Opened -> LogicalClose -> PhysicalClose`
 /// without invoking a listener while the lock is held.
 #[derive(Debug)]
 enum LifecycleEventProgress {
     /// Pool installation completed, but the `Opened` callback has not returned.
     OpenedPending,
     /// `Opened` returned and logical close is the next event.
     WaitingForLogicalClose(SharedConnectionEventListener),
     /// Logical close was reported and physical close is the next event.
     /// The logical-close callback is running outside the lifecycle lock.
     LogicalClosePending(SharedConnectionEventListener),
     /// Logical close returned and physical close is the next event.
     WaitingForPhysicalClose(SharedConnectionEventListener),
     /// No further installed-connection event remains.
     Complete,
-375
+}
-376
 impl LifecycleEventProgress {
     /// Advances callback progress after the `Opened` callback returns.
     fn opened_completed(
         &mut self,
         listener: Option<&SharedConnectionEventListener>,
         close: Option<(LogicalCloseCause, Option<CloseReason>)>,
         physically_closed: bool,
     ) -> Option<(LogicalCloseCause, Option<CloseReason>)> {
         close_cause: Option<LogicalCloseCause>,
     ) -> Option<(SharedConnectionEventListener, LogicalCloseCause)> {
         assert!(
             matches!(self, Self::OpenedPending),
             "connection opened event completed more than once"
         );
         let Some(listener) = listener else {
             *self = Self::Complete;
             return None;
         };
         match (close, physically_closed) {
             (None, _) => {
         match close_cause {
             None => {
                 *self = Self::WaitingForLogicalClose(listener.clone());
                 None
-396
+            }
             (Some((cause, _)), false) => {
                 *self = Self::WaitingForPhysicalClose(listener.clone());
                 Some((cause, None))
--
+            }
             (Some((cause, final_reason)), true) => {
                 *self = Self::Complete;
                 Some((
                     cause,
                     Some(final_reason.expect("physical close completed before H1 classification")),
                 ))
             Some(cause) => {
                 *self = Self::LogicalClosePending(listener.clone());
                 Some((listener.clone(), cause))
-400
+            }
-401
+        }
-402
+    }
-403
     /// Returns the listener when logical close may now be reported.
     fn logical_closed(
         &mut self,
         final_reason: Option<CloseReason>,
         physically_closed: bool,
     ) -> Option<(SharedConnectionEventListener, Option<CloseReason>)> {
     fn logical_closed(&mut self) -> Option<SharedConnectionEventListener> {
         match std::mem::replace(self, Self::Complete) {
             Self::OpenedPending => {
                 *self = Self::OpenedPending;
                 None
-410
+            }
             Self::WaitingForLogicalClose(listener) => {
                 if physically_closed {
                     Some((
                         listener,
                         Some(
                             final_reason
                                 .expect("physical close completed before H1 classification"),
                         ),
                     ))
                 } else {
                     *self = Self::WaitingForPhysicalClose(listener.clone());
                     Some((listener, None))
--
+                }
                 *self = Self::LogicalClosePending(listener.clone());
                 Some(listener)
-+
+            }
             Self::LogicalClosePending(listener) => {
                 *self = Self::LogicalClosePending(listener);
                 None
-418
+            }
             Self::WaitingForPhysicalClose(listener) => {
                 *self = Self::WaitingForPhysicalClose(listener);
                 None
-422
+            }
             Self::Complete => None,
-424
+        }
-425
+    }
-426
     /// Advances callback progress after the logical-close callback returns.
     fn logical_close_completed(
         &mut self,
         physically_closed: bool,
     ) -> Option<SharedConnectionEventListener> {
         match std::mem::replace(self, Self::Complete) {
             Self::LogicalClosePending(listener) if physically_closed => Some(listener),
             Self::LogicalClosePending(listener) => {
                 *self = Self::WaitingForPhysicalClose(listener);
                 None
-+
+            }
             state => {
                 *self = state;
                 panic!("connection logical-close event completed out of order");
-+
+            }
-+
+        }
-+
+    }
-+
     /// Returns the listener when physical close may now be reported.
     fn physical_closed(&mut self) -> Option<SharedConnectionEventListener> {
         match std::mem::replace(self, Self::Complete) {
             Self::OpenedPending => {
                 *self = Self::OpenedPending;
                 None
-451
+            }
             Self::WaitingForLogicalClose(listener) => {
                 *self = Self::WaitingForLogicalClose(listener);
                 None
-455
+            }
             Self::LogicalClosePending(listener) => {
                 *self = Self::LogicalClosePending(listener);
                 None
-+
+            }
             Self::WaitingForPhysicalClose(listener) => Some(listener),
             Self::Complete => None,
-462
+        }
-463
+    }
-464
+}
 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 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>,
         owner_spawner: StdArc<dyn DriverSpawner>,
         stats: Arc<CellConnectionStats>,
     ) -> (Arc<Self>, PhysicalConnectionGuard) {
         stats.physical_connection_started();
         let connection = Arc::new(Self {
             info,
             owner_spawner,
             stats,
             lifecycle: Mutex::new(ConnectionLifecycle {
                 logical: LogicalState::PendingOpen,
                 in_flight: 0,
                 physical_connection_complete: false,
                 counted_close: None,
                 events: LifecycleEventProgress::OpenedPending,
             }),
         });
         let physical = PhysicalConnectionGuard {
             connection: connection.clone(),
             active: true,
         };
         (connection, physical)
-499
+    }
-500
     #[cfg(test)]
     pub(super) fn pending_open_for_test(
         info: Arc<ConnectionInfo>,
     ) -> (Arc<Self>, PhysicalConnectionGuard) {
         Self::pending_open(info, Arc::new(CellConnectionStats::default()))
         Self::pending_open(
             info,
             StdArc::new(TestDriverSpawner),
             Arc::new(CellConnectionStats::default()),
-+
+        )
-+
+    }
-+
     /// Returns the runtime that owns this installed connection.
     pub(super) fn owner_spawner(&self) -> StdArc<dyn DriverSpawner> {
         self.owner_spawner.clone()
-515
+    }
-516
     /// 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>> {

↧ Expand context

         let mut lifecycle = self.lifecycle.lock();
         if !matches!(lifecycle.logical, LogicalState::PendingOpen) {
             return Err(lease);
-528
+        }
         lifecycle.logical = LogicalState::Open { capacity: lease };
         Ok(())
-531
+    }
-532
     /// Creates a connection whose origin has no admission bound.
     ///
     /// 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_for_test(info);
         connection
             .open(None)
             .expect("new unbounded connection could not open");
         (connection, physical)
-543
+    }
-544

@@ -530,581 +608,671 @@

↥ Expand context

         let mut lifecycle = connection.lifecycle.lock();
         if !matches!(lifecycle.logical, LogicalState::Open { .. }) {
             return None;
-584
+        }
         lifecycle.in_flight = lifecycle
             .in_flight
             .checked_add(1)
             .expect("in-flight dispatch count exhausted");
         drop(lifecycle);
-590
         Some(DispatchGuard {
             connection: connection.clone(),
             active: true,
         })
-595
+    }
-596
     /// Completes the `Opened` callback and reports any close that occurred
     /// while it was running.
     ///
     /// A listener may reenter the pool and close this connection from its

     /// `Opened` callback. The lifecycle transition completes immediately, but
     /// its callbacks are delayed until this method can preserve
     /// `Opened`, logical close, and physical close in that order.
     pub(super) fn complete_opened_event(&self, listener: Option<&SharedConnectionEventListener>) {
         let pending_close = {
             let mut lifecycle = self.lifecycle.lock();
             assert!(
                 !matches!(lifecycle.logical, LogicalState::PendingOpen),
                 "connection opened event completed before the connection opened"
             );
             let close = match &lifecycle.logical {
             let close_cause = match &lifecycle.logical {
                 LogicalState::PendingOpen | LogicalState::Open { .. } => None,
                 LogicalState::Closed {
                     cause, disposition, ..
                 } => Some((*cause, disposition.final_reason())),
                 LogicalState::Closed { cause, .. } => Some(*cause),
             };
             let physically_closed = lifecycle.physical_connection_complete;
             lifecycle
                 .events
                 .opened_completed(listener, close, physically_closed)
             lifecycle.events.opened_completed(listener, close_cause)
         };
         let Some(listener) = listener else {
             return;
         };
         if let Some((cause, physical_reason)) = pending_close {
         if let Some((listener, cause)) = pending_close {
             listener.logical_close(&self.info, cause);
             if let Some(reason) = physical_reason {
                 listener.physical_close(&self.info, reason);
--
+            }
             self.complete_logical_close_event();
-+
+        }
-+
+    }
-+
     /// Completes logical-close callback delivery and reports a deferred
     /// physical close.
     fn complete_logical_close_event(&self) {
         let pending_physical_close = {
             let mut lifecycle = self.lifecycle.lock();
             let physically_closed = lifecycle.physical_connection_complete;
             let listener = lifecycle.events.logical_close_completed(physically_closed);
             let final_reason = listener.as_ref().map(|_| match lifecycle.logical {
                 LogicalState::Closed { disposition, .. } => disposition
                     .final_reason()
                     .expect("physical close completed before H1 classification"),
                 LogicalState::PendingOpen | LogicalState::Open { .. } => {
                     panic!("physical close followed an open connection")
-+
+                }
             });
             listener.zip(final_reason)
         };
         if let Some((listener, reason)) = pending_physical_close {
             listener.physical_close(&self.info, reason);
-642
+        }
-643
+    }
-644
     /// 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
     /// logical cause remains unchanged.
     ///
     /// Capacity normally returns after the connection lock is released.

↧ Expand context

     /// HTTP/1 retains it while an accepted exchange may still transfer
     /// upgraded I/O.
     pub(super) fn logical_close(&self, reason: CloseReason) -> bool {
         let cause = LogicalCloseCause::from_reason(reason);
         let (released_capacity, pending_events) = {
             let mut lifecycle = self.lifecycle.lock();
             let previous = std::mem::replace(&mut lifecycle.logical, LogicalState::PendingOpen);
             let (disposition, retained_capacity, released_capacity, was_open) = match previous {
                 LogicalState::PendingOpen => {
                     let disposition = match self.info.protocol() {
                         ConnectionProtocol::Http1 => {
                             CloseDisposition::Http1(H1CloseDisposition::Final(reason))
-664
+                        }
                         ConnectionProtocol::Http2 => CloseDisposition::Http2(reason),
                     };
                     (disposition, None, None, false)
-668
+                }
                 LogicalState::Open { mut capacity } => match self.info.protocol() {
                     ConnectionProtocol::Http1
                         if lifecycle.in_flight != 0

@@ -615,678 +773,826 @@

↥ Expand context

-678
+                        (
                             CloseDisposition::Http1(H1CloseDisposition::AwaitingExchange {
                                 fallback: reason,
                             }),
                             capacity.take(),
                             None,
                             true,
-685
+                        )
-686
+                    }
                     ConnectionProtocol::Http1 => {
                         let retain_for_upgrade = reason == CloseReason::Upgraded
                             && !lifecycle.physical_connection_complete;
                         let retained_capacity =
                             retain_for_upgrade.then(|| capacity.take()).flatten();
-692
+                        (
                             CloseDisposition::Http1(H1CloseDisposition::Final(reason)),
                             retained_capacity,
                             capacity,
                             true,
-697
+                        )

-698
+                    }
                     ConnectionProtocol::Http2 => {
                         (CloseDisposition::Http2(reason), None, capacity, true)
-701
+                    }
                 },
                 closed @ LogicalState::Closed { .. } => {
                     lifecycle.logical = closed;
                     return false;
-706
+                }
             };
             let final_reason = disposition.final_reason();
             if was_open && !lifecycle.physical_connection_complete {
                 let counted_close = match self.info.protocol() {
                     ConnectionProtocol::Http1 if reason == CloseReason::Upgraded => {
                         CountedClose::H1Upgraded
-712
+                    }
                     ConnectionProtocol::Http1 => CountedClose::H1Draining,
                     ConnectionProtocol::Http2 => CountedClose::H2Draining,
                 };
                 assert!(
                     lifecycle.counted_close.is_none(),
                     "connection close was counted more than once"
                 );
                 counted_close.start(&self.stats);
                 lifecycle.counted_close = Some(counted_close);
-722
+            }
             lifecycle.logical = LogicalState::Closed {
                 cause,
                 disposition,
                 retained_capacity,
             };
             let physically_closed = lifecycle.physical_connection_complete;
             let pending_events = lifecycle
                 .events
                 .logical_closed(final_reason, physically_closed);
             let pending_events = lifecycle.events.logical_closed();
             (released_capacity, pending_events)
         };
         drop(released_capacity);
         if let Some((listener, physical_reason)) = pending_events {
         if let Some(listener) = pending_events {
             listener.logical_close(&self.info, cause);
             if let Some(reason) = physical_reason {
                 listener.physical_close(&self.info, reason);
--
+            }
             self.complete_logical_close_event();
-735
+        }
         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
-747
+    }
-748
     /// Finalizes a driver-observed HTTP/1 close from exchange-side evidence.
     ///
     /// A confirmed upgrade retains bounded capacity until root I/O leaves the
     /// client. Every other classification preserves the driver reason and
     /// returns capacity immediately.
     pub(super) fn complete_h1_exchange(&self, exchange_reason: CloseReason) -> bool {
         let (released_capacity, listener, final_reason) = {
         let (released_capacity, final_reason) = {
             let mut lifecycle = self.lifecycle.lock();
             let physical_connection_complete = lifecycle.physical_connection_complete;
             let (released_capacity, final_reason) = match &mut lifecycle.logical {
                 LogicalState::Closed {
                     disposition: CloseDisposition::Http1(state),
                     retained_capacity,
                     ..
                 } => {
                     let H1CloseDisposition::AwaitingExchange { fallback } = *state else {
                         return false;
                     };
                     debug_assert!(
                         !physical_connection_complete,
                         "physical close left HTTP/1 exchange classification pending"
                     );
                     let final_reason = if exchange_reason == CloseReason::Upgraded {
                         CloseReason::Upgraded
                     } else {
                         fallback
                     };
                     *state = H1CloseDisposition::Final(final_reason);
                     let release_capacity =
                         final_reason != CloseReason::Upgraded || physical_connection_complete;
-779
+                    (
                         release_capacity.then(|| retained_capacity.take()).flatten(),
                         final_reason,
-782
+                    )
-783
+                }
                 _ => return false,
             };
             if final_reason == CloseReason::Upgraded && !physical_connection_complete {
                 lifecycle.counted_close = Some(
                     lifecycle
                         .counted_close
                         .expect("upgraded HTTP/1 connection had no draining count")
                         .h1_upgrade(&self.stats),
                 );
-793
+            }
             let listener = if physical_connection_complete {
                 lifecycle.events.physical_closed()
             } else {
                 None
             };
             (released_capacity, listener, final_reason)
             (released_capacity, final_reason)
         };
         drop(released_capacity);
         if let Some(listener) = listener {
             listener.physical_close(&self.info, final_reason);
--
+        }
         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(),
             close_reason = ?final_reason,
             exchange_reason = ?exchange_reason,
             "HTTP/1 connection close classification completed"
         );

↧ Expand context

         true
-808
+    }
-809
     /// Removes one dispatch previously committed by [`Self::try_commit_dispatch`].
     fn release_dispatch(&self) {
         let complete_reason = {
             let mut lifecycle = self.lifecycle.lock();
             lifecycle.in_flight = lifecycle
                 .in_flight
                 .checked_sub(1)
                 .expect("completed a dispatch that was not in flight");
             if lifecycle.in_flight == 0 {
                 match lifecycle.logical {
                     LogicalState::Closed {
                         disposition:
                             CloseDisposition::Http1(H1CloseDisposition::AwaitingExchange { fallback }),
                         ..
                     } => Some(fallback),
                     _ => None,
-826
+                }

@@ -1123,1176 +1183,1237 @@

↥ Expand context

         let events = ConnectionEvents::new(
             Some(SharedConnectionEventListener::new({
                 let stats = stats.clone();
                 let observed = observed.clone();
                 move |event: &ConnectionEvent<'_>| {
                     let kind = match event {
                         ConnectionEvent::Opened(_) => EventKind::Opened,
                         ConnectionEvent::LogicalClose(_) => EventKind::LogicalClose,
                         ConnectionEvent::PhysicalClose(_) => EventKind::PhysicalClose,
                         ConnectionEvent::EstablishmentFailed(failed) => {
                             panic!("installed connection failed: {failed:?}")
-1187
+                        }
                     };
                     let snapshot = stats.snapshot(0, 0, 0, 0, 0);
                     observed.lock().unwrap().push((
                         kind,
                         snapshot.establishing_connections(),
                         snapshot.h1().draining(),
                         snapshot.physically_live_connections(),
                     ));

-1196
+                }
             })),
             SharedTimeSource::default(),
         );
         let mut establishment = events.establishment_started(
             test_info(1).origin(),
             PartitionId::from_index(0),
             stats.clone(),
         );
         establishment.protocol_selected(ConnectionProtocol::Http1);
         let (connection, physical) = ConnectionState::pending_open(test_info(1), stats);
         let (connection, physical) =
             ConnectionState::pending_open(test_info(1), StdArc::new(TestDriverSpawner), stats);
         connection.open(None).unwrap();
-1209
         establishment.opened(&connection);
         assert!(connection.logical_close(CloseReason::ProtocolClosed));
         physical.release();
-1213
         assert_eq!(
             &[
                 (EventKind::Opened, 0, 0, 1),
                 (EventKind::LogicalClose, 0, 1, 1),

↧ Expand context

                 (EventKind::PhysicalClose, 0, 0, 0),
             ],
             observed.lock().unwrap().as_slice()
         );
-1222
+    }
-1223
     #[test]
     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);
-1229
         assert!(connection.logical_close(CloseReason::Reclaimed));
         assert!(!connection.logical_close(CloseReason::PoolDropped));
         assert_eq!(1, origin.available_capacity_for_test());
         assert!(!connection.probe().physical_connection_complete);
         assert_eq!(
             Some(CloseReason::Reclaimed),
             connection.probe().close_reason
         );

@@ -1416,1470 +1541,1721 @@

↥ Expand context

         assert!(!connection.probe().physical_connection_complete);
         physical.release();
         assert!(connection.probe().physical_connection_complete);
-1473
+    }
-1474
     #[test]
     fn connection_state_retains_immutable_connection_info() {
         #[derive(Clone, Debug, Eq, PartialEq)]
         struct ConnectorMarker(&'static str);
-1479
         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),
             ConnectionProtocol::Http1,
             Connected::new()
                 .proxy(true)
                 .extra(ConnectorMarker("connector-extra")),

         );
         let (connection, _physical) = ConnectionState::unbounded(info);
-1492
         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!(ConnectionProtocol::Http1, connection.info().protocol());
         assert_eq!(None, connection.info().local_addr());
         assert_eq!(None, connection.info().remote_addr());
         assert_eq!(ConnectPath::ForwardProxy, connection.info().connect_path());
         assert_eq!(None, connection.info().establishment());
         let mut extensions = Extensions::new();
         connection.info().apply_connector_extras(&mut extensions);
         assert_eq!(
             Some(&ConnectorMarker("connector-extra")),
             extensions.get::<ConnectorMarker>()
         );
-1507
+    }
-1508
     #[test]
     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.probe().physical_connection_complete);
-1515
         drop(io);
-1517
         assert!(connection.probe().physical_connection_complete);
-1519
+    }
-1520
+}
-1521
 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use crate::client::pool::admission::OriginAdmission;
     use crate::client::pool::events::{
         ConnectionEvent, ConnectionEvents, SharedConnectionEventListener,
     };
     use aws_smithy_async::time::SharedTimeSource;
     use loom::sync::atomic::{AtomicUsize, Ordering};
     use std::num::NonZeroUsize;
-1532
     fn test_info(id: u64) -> Arc<ConnectionInfo> {
         ConnectionInfo::for_test(ConnectionId::new(id), PartitionId::from_index(0))
-1535
+    }
-1536
     /// Races request dispatch commitment with logical close.
     ///
     /// Dispatch either commits before close and drains afterward, or close
     /// rejects it without incrementing the in-flight count.
     #[test]
     fn dispatch_commit_linearizes_against_close() {
         loom::model(|| {
             let (connection, _physical) = ConnectionState::unbounded(test_info(1));
-1545
             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)
             });
-1554
             let dispatch = dispatch.join().unwrap();
             assert!(close.join().unwrap());
             assert!(ConnectionState::try_commit_dispatch(&connection).is_none());
             match dispatch {
                 Some(dispatch) => {
                     let probe = connection.probe();
                     assert_eq!(None, probe.close_reason);
                     assert!(probe.awaiting_h1_exchange);
                     assert_eq!(1, probe.in_flight);
                     drop(dispatch);
                     let probe = connection.probe();
                     assert_eq!(Some(CloseReason::ProtocolClosed), probe.close_reason);
                     assert!(!probe.awaiting_h1_exchange);
                     assert_eq!(0, probe.in_flight);
-1569
+                }
                 None => {
                     let probe = connection.probe();
                     assert_eq!(Some(CloseReason::ProtocolClosed), probe.close_reason);
                     assert!(!probe.awaiting_h1_exchange);
                     assert_eq!(0, probe.in_flight);
-1575
+                }
-1576
+            }
         });
-1578
+    }
-1579
     /// Races two independent logical-close signals for one bounded connection.
     ///
     /// Exactly one reason becomes authoritative and its capacity lease returns once.
     #[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)
             });
-1596
             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.probe().close_reason.unwrap();
             assert!(matches!(
                 reason,
                 CloseReason::Poisoned | CloseReason::PoolDropped
             ));
         });
-1607
+    }
-+
     /// Races HTTP/1 upgrade classification with physical connection completion.
     ///
     /// The winning transition determines the final reason while both paths
     /// converge on one capacity return and zero retained lifetime counts.
     #[test]
     fn h1_upgrade_classification_linearizes_against_physical_completion() {
         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 dispatch = ConnectionState::try_commit_dispatch(&connection)
                 .expect("open HTTP/1 connection rejected dispatch");
             assert!(connection.logical_close(CloseReason::ProtocolClosed));
-+
             let classify_connection = connection.clone();
             let classify = loom::thread::spawn(move || {
                 classify_connection.complete_h1_exchange(CloseReason::Upgraded)
             });
             let complete_physical = loom::thread::spawn(move || physical.release());
-+
             let classified_as_upgrade = classify.join().unwrap();
             complete_physical.join().unwrap();
             dispatch.release();
-+
             let probe = connection.probe();
             assert_eq!(
                 Some(if classified_as_upgrade {
                     CloseReason::Upgraded
                 } else {
                     CloseReason::ProtocolClosed
                 }),
                 probe.close_reason
             );
             assert!(!probe.awaiting_h1_exchange);
             assert_eq!(0, probe.in_flight);
             assert!(probe.physical_connection_complete);
             assert_eq!(1, origin.available_capacity_for_test());
-+
             let stats = connection.stats.snapshot(0, 0, 0, 0, 0);
             assert_eq!(0, stats.h1().draining());
             assert_eq!(0, stats.h1().upgraded());
             assert_eq!(0, stats.physically_live_connections());
         });
-+
+    }
-+
     /// Races `Opened`, accepted HTTP/1 completion, and both close transitions.
     ///
     /// The production lifecycle must report each event once and preserve
     /// `Opened -> LogicalClose -> PhysicalClose` for every interleaving.
     #[test]
     fn installed_events_remain_ordered_across_concurrent_close() {
         let mut model = loom::model::Builder::new();
         model.preemption_bound.get_or_insert(3);
         model.check(|| {
             let event_sequence = Arc::new(AtomicUsize::new(0));
             let observed = event_sequence.clone();
             let events = ConnectionEvents::new(
                 Some(SharedConnectionEventListener::new(
                     move |event: &ConnectionEvent<'_>| {
                         let code = match event {
                             ConnectionEvent::Opened(_) => 1,
                             ConnectionEvent::LogicalClose(_) => 2,
                             ConnectionEvent::PhysicalClose(_) => 3,
                             ConnectionEvent::EstablishmentFailed(_) => 4,
                         };
                         observed
                             .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |sequence| {
                                 Some(sequence * 4 + code)
                             })
                             .expect("event sequence update is infallible");
                     },
                 )),
                 SharedTimeSource::default(),
             );
             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 dispatch = ConnectionState::try_commit_dispatch(&connection)
                 .expect("open HTTP/1 connection rejected dispatch");
             let establishment = events.establishment_started(
                 connection.info().origin(),
                 connection.owner_partition(),
                 connection.stats.clone(),
             );
-+
             let opened_connection = connection.clone();
             let opened = loom::thread::spawn(move || establishment.opened(&opened_connection));
             let close_connection = connection.clone();
             let close = loom::thread::spawn(move || {
                 close_connection.logical_close(CloseReason::ProtocolClosed)
             });
             let complete_physical = loom::thread::spawn(move || physical.release());
             let complete_dispatch = loom::thread::spawn(move || dispatch.release());
-+
             opened.join().unwrap();
             assert!(close.join().unwrap());
             complete_physical.join().unwrap();
             complete_dispatch.join().unwrap();
-+
             assert_eq!(
 ,
                 event_sequence.load(Ordering::SeqCst),
                 "installed connection events were missing, duplicated, or reordered"
             );
             assert_eq!(1, origin.available_capacity_for_test());
             let probe = connection.probe();
             assert_eq!(Some(CloseReason::ProtocolClosed), probe.close_reason);
             assert!(!probe.awaiting_h1_exchange);
             assert_eq!(0, probe.in_flight);
             assert!(probe.physical_connection_complete);
         });
-+
+    }
-1721
+}

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

@@ -10,10 +152,284 @@

↥ Expand context

 //!
 //! 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::H1DispatchOutcome;

 use self::h2::H2DispatchOutcome;
 use super::admission::ProtocolRequirement;
 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_async::time::SharedTimeSource;
 use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
 use aws_smithy_runtime_api::client::http::telemetry::{
     CaptureHttpAttemptTelemetry, ConnectionAcquisitionTelemetry, ConnectionUsage,
 };
 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;
 use std::time::SystemTime;
-51
 /// Replacement selections allowed after the initial HTTP/2 dispatch attempt.
 const MAX_H2_REPLACEMENTS: usize = 2;
-54
 /// 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>,
     /// Inputs needed to start request-attempt observation at acquisition.
     attempt_telemetry: Option<AttemptTelemetryInput>,
-65
+}
-66
 impl RequestOptions {
     /// Creates operation settings for one pool request.
     pub(super) fn new(connect_timeout: Option<TransportTimeout>) -> Self {
         Self { connect_timeout }
     pub(super) fn new(
         connect_timeout: Option<TransportTimeout>,
         attempt_telemetry: Option<AttemptTelemetryInput>,
     ) -> Self {
         Self {
             connect_timeout,
             attempt_telemetry,
-+
+        }
-+
+    }
-+
+}
-+
 /// Capture and operation runtime clock paired before pool dispatch.
 #[derive(Clone, Debug)]
 pub(super) struct AttemptTelemetryInput {
     capture: CaptureHttpAttemptTelemetry,
     time_source: SharedTimeSource,
-+
+}
-+
 impl AttemptTelemetryInput {
     /// Pairs the request capture with the clock that measures its attempt.
     pub(super) fn new(capture: CaptureHttpAttemptTelemetry, time_source: SharedTimeSource) -> Self {
         Self {
             capture,
             time_source,
-+
+        }
-94
+    }
-95
+}
-96
 /// 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>,
     /// Request-attempt observation retained until Hyper accepts a connection.
     pub(super) attempt_telemetry: Option<AttemptTelemetry>,
-+
+}
-+
 /// Request-scoped capture and acquisition start time.
 ///
 /// This value exists only when the request carries a telemetry capture, so an
 /// unobserved request does not read the clock.
 #[derive(Clone)]
 pub(super) struct AttemptTelemetry {
     /// Shared attempt capture installed on the Smithy request.
     capture: CaptureHttpAttemptTelemetry,
     /// Time at which this request entered pool acquisition.
     acquisition_started_at: SystemTime,
     /// Operation runtime clock used for this request attempt.
     time_source: SharedTimeSource,
-+
+}
-+
 impl AttemptTelemetry {
     /// Starts connection acquisition using the operation runtime clock.
     fn start(input: AttemptTelemetryInput) -> Self {
         let AttemptTelemetryInput {
             capture,
             time_source,
         } = input;
         let acquisition_started_at = time_source.now();
         Self {
             capture,
             acquisition_started_at,
             time_source,
-+
+        }
-+
+    }
-+
     /// Freezes one selected connection when the protocol accepts the request.
     fn freeze_selection(
         &self,
         reused: bool,
         metadata: Option<&ConnectionMetadata>,
     ) -> FrozenConnectionSelection {
         FrozenConnectionSelection {
             capture: self.capture.clone(),
             acquisition_started_at: self.acquisition_started_at,
             acquisition_completed_at: self.time_source.now(),
             usage: if reused {
                 ConnectionUsage::Reused
             } else {
                 ConnectionUsage::Fresh
             },
             metadata: metadata
                 .expect("attempt telemetry requires selected connection metadata")
                 .clone(),
-+
+        }
-+
+    }
-+
+}
-+
 /// Selected connection facts held until the request cannot return to acquisition.
 pub(super) struct FrozenConnectionSelection {
     capture: CaptureHttpAttemptTelemetry,
     acquisition_started_at: SystemTime,
     acquisition_completed_at: SystemTime,
     usage: ConnectionUsage,
     metadata: ConnectionMetadata,
-+
+}
-+
 impl FrozenConnectionSelection {
     /// Commits the selected connection and any valid acquisition interval.
     pub(super) fn record(self) {
         let acquisition = ConnectionAcquisitionTelemetry::from_interval(
             self.acquisition_started_at,
             self.acquisition_completed_at,
             self.usage,
         );
         self.capture
             .record_connection_selection(acquisition, self.metadata);
-+
+    }
-+
+}
-+
 impl AcquisitionContext {
     fn captures_connection_selection(&self) -> bool {
         self.attempt_telemetry.is_some()
-+
+    }
-+
     /// Freezes the selected connection when Hyper may retain the request.
     ///
     /// Protocol dispatch records the resulting telemetry only after the request
     /// can no longer be returned to acquisition.
     pub(super) fn freeze_connection_selection(
         &self,
         reused: bool,
         metadata: Option<&ConnectionMetadata>,
     ) -> Option<FrozenConnectionSelection> {
         self.attempt_telemetry
             .as_ref()
             .map(|telemetry| telemetry.freeze_selection(reused, metadata))
-+
+    }
-211
+}
-212
 /// Resolves request-scoped pool state and runs acquisition through dispatch.
 pub(super) async fn send(
     pool: &ConnectionPool,
     partition: Arc<PartitionState>,
     request: Request<SdkBody>,
     options: RequestOptions,
 ) -> Result<Response<SdkBody>, ConnectorError> {
     validate_request_before_acquisition(&request)
         .map_err(|error| ConnectorError::user(error.into()))?;
-222
     let RequestOptions {
         connect_timeout,
         attempt_telemetry,
     } = options;
     let attempt_telemetry = attempt_telemetry.map(AttemptTelemetry::start);
     let absolute_uri = request.uri().clone();
     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"
     );
-241
     let owner_spawner = partition
         .owner_spawner()
         .map_err(|error| ConnectorError::user(error.into()))?;
     partition.ensure_maintenance_started(owner_spawner.as_ref());
-246
     let context = AcquisitionContext {
         pool: pool.inner.clone(),
         partition,
         cell,
         absolute_uri,
         owner_spawner,
         connect_timeout: options.connect_timeout,
         connect_timeout,
         attempt_telemetry,
     };
-256
     acquire_and_dispatch(context, request).await
-258
+}
-259
 /// 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.

↧ Expand context

     H2(H2Activation),
-266
+}
-267
 /// 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> {
     let requirement = protocol_requirement(&request);
     let mut h2_replacements = H2ReplacementBudget::default();
-279
     loop {
         match acquire_for_dispatch(&context, requirement).await? {
             DispatchTarget::H1(selection) => {
                 match h1::dispatch(&context, request, selection).await? {
                     H1DispatchOutcome::Response(response) => return Ok(response),

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

@@ -1,1 +223,232 @@

↥ 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::{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::future::{poll_fn, Future};
 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 H1DispatchOutcome {
     /// Hyper accepted the request and produced a guarded response.
     Response(Response<SdkBody>),
     /// The selection became stale or Hyper returned the request unsent.
     NotAccepted(Request<SdkBody>),
-36
+}
-37
 /// 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<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
     let connection_capture = 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
         });
         .cloned();
     let captured_metadata = (connection_capture.is_some()
         || context.captures_connection_selection())
     .then(|| connection.info().metadata(close_handle.clone()));
     if let (Some(capture), Some(metadata)) = (connection_capture, &captured_metadata) {
         let captured = metadata.clone();
         capture.set_connection_retriever(move || Some(captured.clone()));
-+
+    }
-63
     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(),
-69
+        )
         .with_connection(metadata));
-71
+    }
-72
     add_host_header(&mut request, &context.absolute_uri)
         .map_err(|error| ConnectorError::user(error.into()))?;
     let connect_path = connection.info().connect_path_inner();
     connect_path.apply_proxy_authorization(request.headers_mut());
     rewrite_h1_request_target(&mut request, connect_path.uses_absolute_form());
-78
     // 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(H1DispatchOutcome::NotAccepted(request));
     };
     let send = selection.sender_mut().hyper_mut().try_send_request(request);
--
     let exchange = selection.into_exchange();
     match send.await {
     let mut send = std::pin::pin!(send);
     // Freeze acquisition time when Hyper first retains the request. Publish it
     // only after a later error can no longer return the request envelope.
     let first = poll_fn(|cx| Poll::Ready(send.as_mut().poll(cx))).await;
     let (result, connection_selection) = match first {
         Poll::Ready(result) => (
             result,
             context.freeze_connection_selection(reused, captured_metadata.as_ref()),
         ),
         Poll::Pending => {
             let selection = context.freeze_connection_selection(reused, captured_metadata.as_ref());
             (send.await, selection)
-+
+        }
     };
     match result {
         Ok(mut response) => {
             if let Some(selection) = connection_selection {
                 selection.record();
-+
+            }
             // 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(H1DispatchOutcome::Response(guard_h1_response(
                 response,
                 request_method,
                 exchange,
                 dispatch,
                 context.owner_spawner.clone(),
             )))
-128
+        }
         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(H1DispatchOutcome::NotAccepted(returned));
-145
+                }
                 let metadata =
                     captured_metadata.unwrap_or_else(|| connection.info().metadata(close_handle));
                 return Err(downcast_error(Box::new(error.into_error())).with_connection(metadata));
-149
+            }
             exchange.retire_connection(CloseReason::IncompleteH1Exchange);
             drop(dispatch);
             if let Some(selection) = connection_selection {
                 selection.record();
-+
+            }
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().metadata(close_handle));
             Err(downcast_error(Box::new(error.into_error())).with_connection(metadata))
-158
+        }
-159
+    }
-160
+}
-161
 /// 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_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.release();
         return Response::from_parts(parts, SdkBody::from_body_1_x(body));
-180
+    }
     let body = H1ResponseBody::new(body, exchange, dispatch, spawner);
     let body = H1ResponseBody::new(body, exchange, dispatch);
     Response::from_parts(parts, SdkBody::from_body_1_x(body))
-183
+}
-184
 /// 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>,
-195
+}
-196
 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 {
     fn new(inner: hyper::body::Incoming, exchange: H1Exchange, dispatch: DispatchGuard) -> Self {
         let mut body = Self {
             inner,
             lifecycle: Some(H1ResponseLifecycle {
                 exchange: Some(exchange),
                 dispatch: Some(dispatch),
                 spawner,
             }),
             lifecycle: Some(H1ResponseLifecycle::new(exchange, dispatch)),
         };
         if body.inner.is_end_stream() {
             body.finish_without_context();
-206
+        }
         body
-208
+    }
-209
     /// 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() {

↧ Expand context

             lifecycle.resolve(Some(cx));
-214
+        }
-215
+    }
-216
     /// Completes response ownership without a task context when already ready.
     fn finish_without_context(&mut self) {
         if let Some(lifecycle) = self.lifecycle.take() {
             lifecycle.resolve(None);
-221
+        }
-222
+    }
-223
     /// 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);
-229
+            }
             drop(lifecycle.dispatch.take());
-231
+        }
-232
+    }

@@ -266,275 +325,344 @@

↥ Expand context

 /// The response body owns this value together with the accepted exchange and
 /// dispatch accounting. End-of-stream polls Hyper readiness: success begins
 /// ordinary sender return, failure retires the connection, and pending
 /// readiness transfers the exchange to [`H1ReadinessTask`] on the
 /// connection-owning runtime. Dropping it before completion drops
 /// [`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>,
-289
+}
-290
 /// 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>,
-302
+}
-303
 impl H1ResponseLifecycle {
     /// Retains response ownership and derives follow-up placement from the connection.
     fn new(exchange: H1Exchange, dispatch: DispatchGuard) -> Self {
         let spawner = exchange.connection().owner_spawner();
         Self {
             exchange: Some(exchange),
             dispatch: Some(dispatch),
             spawner,
-+
+        }
-+
+    }
-+
     /// 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 resolve(mut self, cx: Option<&mut Context<'_>>) {
         let mut exchange = self
             .exchange
             .take()
             .expect("HTTP/1 response lifecycle lost its exchange");

↧ Expand context

         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();
-333
+            }
             Poll::Ready(Err(_)) => {
                 exchange.retire_connection(CloseReason::ProtocolClosed);
-336
+            }
             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),
-343
+                    }
                 }));

@@ -399,418 +531,568 @@

↥ Expand context

         http_1x::HeaderValue::from_str(&host).map_err(RequestDispatchError::InvalidHostHeader)?;
     request.headers_mut().insert(http_1x::header::HOST, value);
     request.extensions_mut().insert(H1HostHeaderInserted);
     Ok(())
-422
+}
-423
 /// 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;
-427
     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");
-433
+        }
         return;
-435
+    }
     if is_proxied {
         return;

-438
+    }
     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");
-442
+}
-443
 #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
 mod tests {
     use super::*;
     use crate::client::connect::ConnectPathInner;
     use crate::client::pool::admission::{OriginAdmission, ProtocolRequirement};
     use crate::client::pool::cell::h1::H1Sender;
     use crate::client::pool::cell::OriginCell;
     use crate::client::pool::cell::{AcquisitionOutcome, AcquisitionStep, OriginCell};
     use crate::client::pool::connection::ConnectionInfo;
     use crate::client::pool::dispatch::RequestOptions;
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::EligibilityGroup;
     use crate::client::pool::registry::PartitionState;
     use crate::client::pool::stats::CellConnectionStats;
     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};
     use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
     use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
     use http_1x::uri::Scheme;
     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::num::NonZeroUsize;
     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;
-481
     #[derive(Debug)]
     struct DroppingSpawner;
-484
     impl DriverSpawner for DroppingSpawner {
         fn spawn(&self, driver: Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-488
+        }
-489
+    }
-490
     #[derive(Clone, Debug)]
     struct TrackingSpawner {
         active: StdArc<AtomicUsize>,
-494
+    }
-495
     impl DriverSpawner for TrackingSpawner {
         fn spawn(&self, driver: Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>) {
             self.active.fetch_add(1, Ordering::SeqCst);
             let active = self.active.clone();
             tokio::spawn(async move {
                 struct ActiveGuard(StdArc<AtomicUsize>);
-502
                 impl Drop for ActiveGuard {
                     fn drop(&mut self) {
                         self.0.fetch_sub(1, Ordering::SeqCst);
-506
+                    }
-507
+                }
-508
                 let _guard = ActiveGuard(active);
                 driver.await;
             });
-512
+        }
-513
+    }
-514
     #[derive(Clone, Debug)]
     struct CountingSpawner {
         submitted: StdArc<AtomicUsize>,
-518
+    }
-519
     impl DriverSpawner for CountingSpawner {
         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             self.submitted.fetch_add(1, Ordering::SeqCst);
             drop(tokio::spawn(driver));
-524
+        }
-525
+    }
-526
     #[derive(Clone, Debug)]
     struct CountingDroppingSpawner {
         submitted: StdArc<AtomicUsize>,
-+
+    }
-+
     impl DriverSpawner for CountingDroppingSpawner {
         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             self.submitted.fetch_add(1, Ordering::SeqCst);
             drop(driver);
-+
+        }
-+
+    }
-+
     #[derive(Clone, Debug)]
     struct PausedConnector {
         started: StdArc<Notify>,
         release: StdArc<Notify>,
         calls: StdArc<AtomicUsize>,
-544
+    }
-545
     impl Service<Uri> for PausedConnector {
         type Response = TokioIo<TcpStream>;
         type Error = io::Error;

↧ Expand context

         type Future =
             Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
-551
         fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
             Poll::Ready(Ok(()))
-554
+        }
-555
         fn call(&mut self, uri: Uri) -> Self::Future {
             let address = uri
                 .authority()
                 .expect("test URI has an authority")
                 .as_str()
                 .to_owned();
             let started = self.started.clone();
             let release = self.release.clone();
             let calls = self.calls.clone();
             Box::pin(async move {
                 calls.fetch_add(1, Ordering::SeqCst);
                 started.notify_one();
                 release.notified().await;

@@ -933,970 +1004,1124 @@

↥ Expand context

     #[tokio::test]
     async fn ready_sender_finishes_without_a_readiness_task() {
         let server = TestServer::start().await;
         let submitted = StdArc::new(AtomicUsize::new(0));
         let pool = ConnectionPool::builder()
             .idle_timeout(None)
             .partitions([Partition::new(
                 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());
-999
         H1ResponseLifecycle {
             exchange: Some(exchange),
             dispatch: Some(dispatch),
             spawner: StdArc::new(CountingSpawner {
                 submitted: submitted.clone(),
             }),
--
+        }
         .resolve(None);
         H1ResponseLifecycle::new(exchange, dispatch).resolve(None);
-1001
         assert_eq!(baseline, submitted.load(Ordering::SeqCst));
-1003
+    }
-1004
     #[test]
     fn response_lifecycle_uses_the_connection_owner_spawner() {
         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 submitted = StdArc::new(AtomicUsize::new(0));
         let owner_spawner: StdArc<dyn DriverSpawner> = StdArc::new(CountingSpawner { submitted });
         let (connection, _physical) = ConnectionState::pending_open(
             ConnectionInfo::for_test(ConnectionId::new(1), PartitionId::ANONYMOUS),
             owner_spawner.clone(),
             Arc::new(CellConnectionStats::default()),
         );
         connection.open(None).unwrap();
         let selection =
             OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
         let dispatch =
             ConnectionState::try_commit_dispatch(&connection).expect("connection should be open");
-+
         let lifecycle = H1ResponseLifecycle::new(selection.into_exchange(), dispatch);
-+
         assert!(StdArc::ptr_eq(&owner_spawner, &lifecycle.spawner));
-+
+    }
-+
     #[test]
     fn borrowed_pending_sender_readiness_uses_the_supplier_runtime() {
         let owner_partition = PartitionId::from_index(1);
         let requester_partition = PartitionId::from_index(2);
         let admission = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
         let origin = OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap();
         let owner_cell = OriginAdmission::register_cell(
             &admission,
             Arc::new(OriginCell::new(
                 owner_partition,
                 origin.clone(),
                 EligibilityGroup::Pool,
                 Some(admission.clone()),
                 None,
             )),
         );
         let requester_cell = OriginAdmission::register_cell(
             &admission,
             Arc::new(OriginCell::new(
                 requester_partition,
                 origin,
                 EligibilityGroup::Pool,
                 Some(admission.clone()),
                 None,
             )),
         );
         let submitted = StdArc::new(AtomicUsize::new(0));
         let owner_spawner: StdArc<dyn DriverSpawner> = StdArc::new(CountingDroppingSpawner {
             submitted: submitted.clone(),
         });
         let (connection, _physical) = ConnectionState::pending_open(
             ConnectionInfo::for_test(ConnectionId::new(1), owner_partition),
             owner_spawner.clone(),
             owner_cell.connection_stats(),
         );
         connection
             .open(Some(OriginAdmission::lease_for_test(&admission)))
             .unwrap();
         OriginCell::insert_idle_h1(&owner_cell, connection.clone(), H1Sender::pending_test(11));
-+
         let waiter =
             OriginCell::register_waiter(&requester_cell, ProtocolRequirement::H1Compatible);
         let selection = match requester_cell
             .poll_waiter(waiter, &mut Context::from_waker(std::task::Waker::noop()))
-+
+        {
             Poll::Ready(AcquisitionStep::Resolved(AcquisitionOutcome::H1(selection))) => selection,
             other => panic!("peer demand did not borrow the pending HTTP/1 sender: {other:?}"),
         };
         assert_eq!(owner_partition, selection.connection().owner_partition());
         let dispatch =
             ConnectionState::try_commit_dispatch(&connection).expect("connection should be open");
-+
         let lifecycle = H1ResponseLifecycle::new(selection.into_exchange(), dispatch);
         assert!(StdArc::ptr_eq(&owner_spawner, &lifecycle.spawner));
         lifecycle.resolve(None);
-+
         assert_eq!(1, submitted.load(Ordering::SeqCst));
         assert_eq!(
             Some(CloseReason::OwnerRuntimeShutdown),
             connection.probe().close_reason
         );
-+
+    }
-+
     #[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 {

↧ Expand context

                 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(),
                     RequestOptions::default(),
-1120
+                )
                 .await
         });
         started.notified().await;
-1124

@@ -1506,1626 +1565,1688 @@

↥ Expand context

         let error = client
             .http_connector(&settings, &components)
             .call(HttpRequest::get("http://example.com/").unwrap())
             .await
             .unwrap_err();
         assert!(error.is_timeout(), "unexpected connector error: {error:?}");
-1632
+    }
-1633
     #[tokio::test]
     async fn read_timeout_covers_handshake_through_response_headers() {
         let pool = super::super::super::builder::Builder::default()
             .build_with_connector(NeverReplies)
             .unwrap();
         let client = Client::new(&pool).unwrap();
         let components = RuntimeComponentsBuilder::for_tests()
             .with_sleep_impl(Some(TokioSleep::new()))
             .build()
             .unwrap();
         let settings = HttpConnectorSettings::builder()
             .read_timeout(std::time::Duration::from_millis(10))

             .build();
-1647
         let error = client
             .http_connector(&settings, &components)
             .call(HttpRequest::get("http://example.com/").unwrap())
             .await
             .unwrap_err();
         assert!(error.is_timeout(), "unexpected connector error: {error:?}");
-1654
+    }
-1655
     /// Proves that a successful HTTP/1 upgrade transfers root I/O to the
     /// response while retaining bounded capacity. The pool may establish a
     /// replacement only after the caller drops the upgraded I/O.
     async fn assert_h1_upgrade_retains_capacity(
         method: Method,
         response_head: &'static [u8],
         expected_status: http_1x::StatusCode,
     ) {
         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 request_received = StdArc::new(Notify::new());

↧ Expand context

         let request_received_for_task = request_received.clone();
         let release_response = StdArc::new(Notify::new());
         let release_response_for_task = release_response.clone();
         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 request_received = request_received_for_task.clone();
                     let release_response = release_response_for_task.clone();
                     tokio::spawn(async move {
                         if !read_request_head(&mut stream).await {
                             return;
-1681
+                        }
                         request_received.notify_one();
                         release_response.notified().await;
                         stream.write_all(response_head).await.unwrap();
                         let mut sink = [0_u8; 32];
                         while stream.read(&mut sink).await.unwrap_or(0) != 0 {}
                     });
                 } else {

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

@@ -1,1 +309,317 @@

↥ Expand context

 /*
  * 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 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 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, H2ResponseGuard, H2UploadGuard,
 };
 use super::super::connection::{CloseReason, ConnectionState};
 use super::{AcquisitionContext, H1HostHeaderInserted};
 use super::{AcquisitionContext, FrozenConnectionSelection, 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 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.
     NotAccepted(Box<H2UnacceptedRequest>),
-45
+}
-46
 /// Request and terminal fallback retained for one replacement selection.
 pub(super) struct H2UnacceptedRequest {
     /// Original request Hyper did not accept.
     request: Request<SdkBody>,
     /// Error returned if the request exhausts its replacement budget.
     terminal_error: ConnectorError,
-53
+}
-54
 impl H2UnacceptedRequest {
     /// Returns the original request and its terminal fallback.
     pub(super) fn into_parts(self: Box<Self>) -> (Request<SdkBody>, ConnectorError) {
         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>,
     /// Response guard transferred to the guarded response body.
     response_guard: H2ResponseGuard,
     /// Whether the generation accepted an earlier request.
     reused: bool,
     /// Selected-connection observation held until the request cannot return.
     connection_selection: Option<FrozenConnectionSelection>,
-80
+}
-81
 /// 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<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
     let connection_capture = 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
         });
         .cloned();
     let captured_metadata = (connection_capture.is_some()
         || context.captures_connection_selection())
     .then(|| connection.info().h2_metadata(close.clone()));
     if let (Some(capture), Some(metadata)) = (connection_capture, &captured_metadata) {
         let captured = metadata.clone();
         capture.set_connection_retriever(move || Some(captured.clone()));
-+
+    }
-106
     let H2DispatchParts {
         mut sender,
         upload,
         response,
     } = activation.take_dispatch_parts();
     if sender.is_closed() {
         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),
         );
-122
+    }
     let Some(dispatch) = ConnectionState::try_commit_dispatch(&connection) else {
         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),
         );
     };
-133
     let body = H2RequestBodyHandle::arm(&mut request, upload);
-135
     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;
-138
     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(response);
             drop(dispatch);
             drop(activation);
             close.close(CloseReason::ProtocolClosed);
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
             resolve_unaccepted_request(
                 returned,
                 reused,
                 UnacceptedStage::ReturnedByHyper,
                 downcast_error(Box::new(error.into_error())).with_connection(metadata),
-156
+            )
-157
+        }
         Poll::Ready(result) => {
             activation.accept(dispatch);
             let connection_selection =
                 context.freeze_connection_selection(reused, captured_metadata.as_ref());
             let sender_closed = sender.is_closed();
             resolve_h2_send(
                 sender_closed,
                 result,
                 H2AcceptedDispatch {
                     connection,
                     close,
                     captured_metadata,
                     response_guard: response,
                     reused,
                     connection_selection,
                 },
-174
+            )
-175
+        }
         Poll::Pending => {
             activation.accept(dispatch);
             let connection_selection =
                 context.freeze_connection_selection(reused, captured_metadata.as_ref());
             let result = send.await;
             let sender_closed = sender.is_closed();
             resolve_h2_send(
                 sender_closed,
                 result,
                 H2AcceptedDispatch {
                     connection,
                     close,
                     captured_metadata,
                     response_guard: response,
                     reused,
                     connection_selection,
                 },
-193
+            )
-194
+        }
-195
+    }
-196
+}
-197
 #[allow(
     clippy::result_large_err,
     reason = "ConnectorError preserves SDK classification and connection metadata"
 )]
 fn resolve_h2_send(
     sender_closed: bool,
     result: Result<
         Response<hyper::body::Incoming>,
         hyper::client::conn::TrySendError<Request<SdkBody>>,
     >,
     accepted: H2AcceptedDispatch,
 ) -> Result<H2DispatchOutcome, ConnectorError> {
     let H2AcceptedDispatch {
         connection,
         close,
         captured_metadata,
         response_guard,
         reused,
         connection_selection,
     } = accepted;
     match result {
         Ok(mut response) => {
             if let Some(selection) = connection_selection {
                 selection.record();
-+
+            }
             connection
                 .info()
                 .apply_connector_extras(response.extensions_mut());
             let (parts, body) = response.into_parts();
             let body = H2ResponseBody::new(body, response_guard);
             Ok(H2DispatchOutcome::Response(Response::from_parts(
                 parts,
                 SdkBody::from_body_1_x(body),
             )))
-228
+        }
         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();
-236
+                }
                 drop(response_guard);
                 close.close(CloseReason::ProtocolClosed);
                 let metadata =
                     captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
                 return resolve_unaccepted_request(
                     request,
                     reused,
                     UnacceptedStage::ReturnedByHyper,
                     downcast_error(Box::new(error.into_error())).with_connection(metadata),
                 );
-247
+            }
             if sender_closed {
                 close.close(CloseReason::ProtocolClosed);
-250
+            }
             drop(response_guard);
             if let Some(selection) = connection_selection {
                 selection.record();
-+
+            }
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
             Err(downcast_error(Box::new(error.into_error())).with_connection(metadata))
-258
+        }
-259
+    }
-260
+}
-261
 /// 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()
-269
+    {
         request.headers_mut().remove(http_1x::header::HOST);
-271
+    }
-272
+}
-273
 /// 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,
-281
+}
-282
 /// 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 resolve_unaccepted_request(
     request: Request<SdkBody>,
     reused: bool,
     stage: UnacceptedStage,
     error: ConnectorError,
 ) -> Result<H2DispatchOutcome, ConnectorError> {
     if stage == UnacceptedStage::BeforeHyper || reused {
         Ok(H2DispatchOutcome::NotAccepted(Box::new(
             H2UnacceptedRequest {
                 request,

↧ Expand context

                 terminal_error: error,
             },
         )))
     } else {
         Err(error)
-303
+    }
-304
+}
-305
 /// A fresh HTTP/2 generation closed before accepting its first request.
 #[derive(Debug)]
 struct H2ConnectionClosedBeforeDispatch;
-309
 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")
-313
+    }
-314
+}
-315
 impl std::error::Error for H2ConnectionClosedBeforeDispatch {}
-317

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

@@ -56,56 +154,155 @@

↥ Expand context

                 origin_scheme = %connection.info().origin().scheme(),
                 origin_host = connection.info().origin().host(),
                 origin_port = ?connection.info().origin().port(),
                 "HTTP/1 connection established"
             );
-61
+        }
         Err(error) => tracing::debug!(
             request_partition = ?request_partition,
             connection_partition = ?connection_cell.id().partition(),
             origin_scheme = %connection_cell.id().origin().scheme(),
             origin_host = connection_cell.id().origin().host(),
             origin_port = ?connection_cell.id().origin().port(),
             error = ?error,
             "HTTP/1 connection establishment failed"
         ),
-71
+    }
     result
-73
+}
-74
 /// Handshakes and installs one already connected HTTP/1 transport.

 async fn run_h1_handshake(
     context: AcquisitionContext,
     permit: EstablishmentPermit,
     transport: ConnectedTransport,
     mut establishment: ConnectionEstablishment,
 ) -> Result<H1Selection, ConnectorError> {
     let AcquisitionContext {
         pool,
         partition,
         cell,
         absolute_uri: _,
         owner_spawner,
         connect_timeout: _,
         ..
     } = context;
-89
     let id = match next_connection_id(&pool) {
         Ok(id) => id,
         Err(error) => {
             let error = ConnectorError::other(error.into(), None);
             establishment.failed(&error);
             return Err(error);
-96
+        }
     };
     let info = ConnectionInfo::new(
         id,
         cell.id().origin().clone(),
         partition.id(),
         ConnectionProtocol::Http1,
         transport.metadata,
     );
     let (connection, physical) = ConnectionState::pending_open(info, cell.connection_stats());
     let (connection, physical) =
         ConnectionState::pending_open(info, owner_spawner, cell.connection_stats());
     let io = ConnectionIo::new(transport.io, physical);
-108
     establishment.protocol_handshake_started();
     let (sender, driver) = match hyper::client::conn::http1::Builder::new()
         .handshake::<_, SdkBody>(io)
         .await
-113
+    {
         Ok(established) => established,
         Err(error) => {
             let error = downcast_error(Box::new(error));
             establishment.protocol_handshake_failed();
             connection.logical_close(CloseReason::ProtocolClosed);
             establishment.failed(&error);
             return Err(error);
-121
+        }
     };
     establishment.protocol_handshake_completed();
-124
     if let Err(lease) = connection.open(permit.into_lease()) {
         drop(lease);
         connection.logical_close(CloseReason::ProtocolClosed);
         let error = ConnectorError::io("HTTP/1 connection closed before installation".into());
         establishment.failed(&error);
         return Err(error);
-131
+    }
     establishment.installed(&connection);
-133
     let selection =
         OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::from_hyper(sender));
     let driver_guard = H1DriverGuard::new(H1CloseHandle::new(&cell, &connection));
     let driver_info = connection.info().clone();
     owner_spawner.spawn(Box::pin(async move {
     connection.owner_spawner().spawn(Box::pin(async move {
         let result = driver.with_upgrades().await;
         if let Err(error) = result {
             tracing::debug!(
                 connection_id = %driver_info.id(),
                 connection_partition = ?driver_info.owner_partition(),
                 origin_scheme = %driver_info.origin().scheme(),
                 origin_host = driver_info.origin().host(),
                 origin_port = ?driver_info.origin().port(),
                 error = ?error,
                 "HTTP/1 connection driver failed"

↧ Expand context

             );
-150
+        }
         driver_guard.protocol_closed();
     }));
     establishment.opened(&connection);
     Ok(selection)
-155
+}

Pages: 1 2 3 4

             );
-316
+        }
-317
+    }
-318
+}