AWS SDK

rev. 1ed3d5371c342b375a8e2197cd2254b0dc08a51c

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,

@@ -423,457 +508,553 @@

↥ Expand context

                     };
                     return None;
-459
+                }
                 let reason = match &self.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")
-466
+                    }
                 };
                 Some((listener, ReadyLifecycleEvent::PhysicalClose(reason)))
-469
+            }
             LifecycleEventProgress::Complete => None,
-471
+        }
-472
+    }
-473
+}
 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)
-508
+    }
-509
     #[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()
-524
+    }
-525
     /// 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);
-537
+        }
         lifecycle.logical = LogicalState::Open { capacity: lease };
         Ok(())
-540
+    }
-541
     /// 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)
-552
+    }
-553

@@ -1130,1175 +1190,1236 @@

↥ 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:?}")
-1186
+                        }
                     };
                     let snapshot = stats.snapshot(0, 0, 0, 0, 0);
                     observed.lock().unwrap().push((
                         kind,
                         snapshot.establishing_connections(),
                         snapshot.h1().draining(),
                         snapshot.physically_live_connections(),
                     ));

-1195
+                }
             })),
             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();
-1208
         establishment.opened(&connection);
         assert!(connection.logical_close(CloseReason::ProtocolClosed));
         physical.release();
-1212
         assert_eq!(
             &[
                 (EventKind::Opened, 0, 0, 1),
                 (EventKind::LogicalClose, 0, 1, 1),

↧ Expand context

                 (EventKind::PhysicalClose, 0, 0, 0),
             ],
             observed.lock().unwrap().as_slice()
         );
-1221
+    }
-1222
     #[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);
-1228
         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
         );

@@ -1444,1490 +1634,1741 @@

↥ Expand context

         assert!(!connection.probe().physical_connection_complete);
         physical.release();
         assert!(connection.probe().physical_connection_complete);
-1493
+    }
-1494
     #[test]
     fn connection_state_retains_immutable_connection_info() {
         #[derive(Clone, Debug, Eq, PartialEq)]
         struct ConnectorMarker(&'static str);
-1499
         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);
-1512
         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>()
         );
-1527
+    }
-1528
     #[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);
-1535
         drop(io);
-1537
         assert!(connection.probe().physical_connection_complete);
-1539
+    }
-1540
+}
-1541
 #[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;
-1552
     fn test_info(id: u64) -> Arc<ConnectionInfo> {
         ConnectionInfo::for_test(ConnectionId::new(id), PartitionId::from_index(0))
-1555
+    }
-1556
     /// 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));
-1565
             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)
             });
-1574
             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);
-1589
+                }
                 None => {
                     let probe = connection.probe();
                     assert_eq!(Some(CloseReason::ProtocolClosed), probe.close_reason);
                     assert!(!probe.awaiting_h1_exchange);
                     assert_eq!(0, probe.in_flight);
-1595
+                }
-1596
+            }
         });
-1598
+    }
-1599
     /// 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)
             });
-1616
             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
             ));
         });
-1627
+    }
-1628
     /// Races `Opened` completion with logical and physical close.
     /// 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.
     ///
     /// Each installed event must be delivered once in
     /// `Opened -> LogicalClose -> PhysicalClose` order.
     #[test]
     fn installed_events_remain_ordered_across_concurrent_close() {
         loom::model(|| {
         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(),
             );
-1713
             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::PoolDropped)
                 close_connection.logical_close(CloseReason::ProtocolClosed)
             });
             let complete_physical = loom::thread::spawn(move || physical.release());
             let complete_dispatch = loom::thread::spawn(move || dispatch.release());
-1722
             opened.join().unwrap();
             assert!(close.join().unwrap());
             complete_physical.join().unwrap();
             complete_dispatch.join().unwrap();
-1727
             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::PoolDropped), probe.close_reason);
             assert_eq!(Some(CloseReason::ProtocolClosed), probe.close_reason);
             assert!(!probe.awaiting_h1_exchange);
             assert_eq!(0, probe.in_flight);
             assert!(probe.physical_connection_complete);
         });
-1740
+    }
-1741
+}

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 +319,327 @@

↥ 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::connect::ConnectPathInner;
 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};
-35
 /// 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>),
-46
+}
-47
 /// 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,
-54
+}
-55
 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)
-64
+    }
-65
+}
-66
 /// 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>,
-81
+}
-82
 /// 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> {
     let connection = activation.connection().clone();
     prepare_h2_request(
         &mut request,
         &context.absolute_uri,
         connection.info().connect_path_inner(),
     );
     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()));
-+
+    }
-111
     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),
         );
-127
+    }
     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),
         );
     };
-138
     let body = H2RequestBodyHandle::arm(&mut request, upload);
-140
     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;
-143
     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),
-161
+            )
-162
+        }
         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,
                 },
-179
+            )
-180
+        }
         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,
                 },
-198
+            )
-199
+        }
-200
+    }
-201
+}
-202
 #[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),
             )))
-233
+        }
         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();
-241
+                }
                 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),
                 );
-252
+            }
             if sender_closed {
                 close.close(CloseReason::ProtocolClosed);
-255
+            }
             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))
-263
+        }
-264
+    }
-265
+}
-266
 /// Restores the absolute URI and applies connection-path request metadata.
 fn prepare_h2_request(
     request: &mut Request<SdkBody>,
     absolute_uri: &http_1x::Uri,
     connect_path: &ConnectPathInner,
 ) {
     *request.uri_mut() = absolute_uri.clone();
     if request
         .extensions_mut()
         .remove::<H1HostHeaderInserted>()
         .is_some()
-278
+    {
         request.headers_mut().remove(http_1x::header::HOST);
-280
+    }
     connect_path.apply_proxy_authorization(request.headers_mut());
-282
+}
-283
 /// 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,
-291
+}
-292
 /// 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)
-313
+    }
-314
+}
-315
 /// A fresh HTTP/2 generation closed before accepting its first request.
 #[derive(Debug)]
 struct H2ConnectionClosedBeforeDispatch;
-319
 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")
-323
+    }
-324
+}
-325
 impl std::error::Error for H2ConnectionClosedBeforeDispatch {}
-327

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
+}

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

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

↥ Expand context

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

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

↧ Expand context

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

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

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

↥ Expand context

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

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

↧ Expand context

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

Pages: 1 2 3 4

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