AWS SDK

rev. 7a81c738582b03212f62160d093a63f8547fc0f5

Files changed:

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

@@ -0,1 +0,642 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-+
 //! HTTP/2 dispatch and two-ended request completion.
 //!
 //! [`H2Activation`] arrives with a prospective generation lease and a transient
 //! sender cloned from the connection-owning cell. Dispatch first checks sender
 //! and logical-connection state, then polls Hyper once. A request returned from
 //! that poll was not accepted and may re-enter protocol acquisition.
 //!
 //! After Hyper accepts the request, the activation becomes an accepted lease.
 //! [`H2RequestBody`] owns its send endpoint and [`H2ResponseBody`] owns its
 //! receive endpoint. Either may finish first; the generation request count is
 //! released only after both finish. A returned request re-arms the existing
 //! body wrapper rather than nesting wrappers around the original body.
-+
 use super::super::cell::h2::{H2Activation, H2CloseHandle, H2DispatchParts, H2LeaseEndpoint};
 use super::super::connection::ConnectionState;
 use super::super::ConnectionPool;
 use super::{AcquisitionContext, H1HostHeaderInserted};
 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};
-+
 /// Result of one checked HTTP/2 dispatch attempt.
 #[allow(
     clippy::large_enum_variant,
     reason = "boxing the successful response would allocate on every HTTP/2 request"
 )]
 pub(super) enum H2DispatchResult {
     /// Hyper accepted the request and produced a guarded response.
     Response(Response<SdkBody>),
     /// Hyper did not accept the request, so it may re-enter acquisition.
     Reacquire(Box<H2Reacquisition>),
-+
+}
-+
 /// Request and terminal fallback retained for one replacement selection.
 pub(super) struct H2Reacquisition {
     /// Original request Hyper did not accept.
     request: Request<SdkBody>,
     /// Error returned if the request exhausts its replacement budget.
     error: ConnectorError,
-+
+}
-+
 impl H2Reacquisition {
     /// Returns the original request and its terminal fallback.
     pub(super) fn into_parts(self: Box<Self>) -> (Request<SdkBody>, ConnectorError) {
         let Self { request, error } = *self;
         (request, error)
-+
+    }
-+
+}
-+
 /// State retained after Hyper accepts one request.
 struct H2AcceptedDispatch {
     /// Connection retained through response creation.
     connection: Arc<ConnectionState>,
     /// Generation close authority used for connection-wide failures.
     close: H2CloseHandle,
     /// Metadata captured before the request moved into Hyper.
     captured_metadata: Option<aws_smithy_runtime_api::client::connection::ConnectionMetadata>,
     /// Receive endpoint transferred to the guarded response body.
     receive_endpoint: H2LeaseEndpoint,
     /// Whether the generation accepted an earlier request.
     reused: bool,
-+
+}
-+
 impl ConnectionPool {
     /// Dispatches one request through a prospective H2 generation lease.
     ///
     /// Pool-side staleness reacquires before Hyper sees the request. Once
     /// Hyper returns an envelope, only a reused generation may reacquire.
     pub(super) async fn dispatch_h2(
         &self,
         context: &AcquisitionContext,
         mut request: Request<SdkBody>,
         mut activation: H2Activation,
     ) -> Result<H2DispatchResult, ConnectorError> {
         prepare_h2_request(&mut request, &context.absolute_uri);
         let connection = activation.connection().clone();
         let reused = activation.is_reused();
         let close = activation.close_handle();
         let captured_metadata = request
             .extensions()
             .get::<CaptureSmithyConnection>()
             .cloned()
             .map(|capture| {
                 let metadata = connection.info().h2_metadata(close.clone());
                 let captured = metadata.clone();
                 capture.set_connection_retriever(move || Some(captured.clone()));
                 metadata
             });
-+
         let H2DispatchParts {
             mut sender,
             send_endpoint,
             receive_endpoint,
         } = activation.take_dispatch_parts();
         if sender.is_closed() {
             close.close(super::super::connection::CloseReason::ProtocolClosed);
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
             return finish_unaccepted_request(
                 request,
                 reused,
                 UnacceptedStage::BeforeHyper,
                 ConnectorError::other(H2ConnectionClosedBeforeDispatch.into(), None)
                     .with_connection(metadata),
             );
-+
+        }
         let Some(dispatch) = ConnectionState::try_commit_dispatch(&connection) else {
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
             return finish_unaccepted_request(
                 request,
                 reused,
                 UnacceptedStage::BeforeHyper,
                 ConnectorError::other(H2ConnectionClosedBeforeDispatch.into(), None)
                     .with_connection(metadata),
             );
         };
-+
         let body = H2RequestBodyHandle::arm(&mut request, send_endpoint);
-+
         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;
-+
         match first {
             Poll::Ready(Err(mut error)) if error.message().is_some() => {
                 let returned = error
                     .take_message()
                     .expect("checked returned request disappeared");
                 body.clear();
                 drop(receive_endpoint);
                 drop(dispatch);
                 drop(activation);
                 close.close(super::super::connection::CloseReason::ProtocolClosed);
                 let metadata =
                     captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
                 finish_unaccepted_request(
                     returned,
                     reused,
                     UnacceptedStage::ReturnedByHyper,
                     super::super::super::downcast_error(Box::new(error.into_error()))
                         .with_connection(metadata),
-+
+                )
-+
+            }
             Poll::Ready(result) => {
                 activation.accept(dispatch);
                 let sender_closed = sender.is_closed();
                 self.finish_h2_send(
                     sender_closed,
                     result,
                     H2AcceptedDispatch {
                         connection,
                         close,
                         captured_metadata,
                         receive_endpoint,
                         reused,
                     },
-+
+                )
-+
+            }
             Poll::Pending => {
                 activation.accept(dispatch);
                 let result = send.await;
                 let sender_closed = sender.is_closed();
                 self.finish_h2_send(
                     sender_closed,
                     result,
                     H2AcceptedDispatch {
                         connection,
                         close,
                         captured_metadata,
                         receive_endpoint,
                         reused,
                     },
-+
+                )
-+
+            }
-+
+        }
-+
+    }
-+
     #[allow(
         clippy::result_large_err,
         reason = "ConnectorError preserves SDK classification and connection metadata"
     )]
     fn finish_h2_send(
         &self,
         sender_closed: bool,
         result: Result<
             Response<hyper::body::Incoming>,
             hyper::client::conn::TrySendError<Request<SdkBody>>,
         >,
         accepted: H2AcceptedDispatch,
     ) -> Result<H2DispatchResult, ConnectorError> {
         let H2AcceptedDispatch {
             connection,
             close,
             captured_metadata,
             receive_endpoint,
             reused,
         } = accepted;
         match result {
             Ok(mut response) => {
                 connection
                     .info()
                     .apply_connector_extras(response.extensions_mut());
                 let (parts, body) = response.into_parts();
                 let body = H2ResponseBody::new(body, receive_endpoint);
                 Ok(H2DispatchResult::Response(Response::from_parts(
                     parts,
                     SdkBody::from_body_1_x(body),
                 )))
-+
+            }
             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();
-+
+                    }
                     drop(receive_endpoint);
                     close.close(super::super::connection::CloseReason::ProtocolClosed);
                     let metadata =
                         captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
                     return finish_unaccepted_request(
                         request,
                         reused,
                         UnacceptedStage::ReturnedByHyper,
                         super::super::super::downcast_error(Box::new(error.into_error()))
                             .with_connection(metadata),
                     );
-+
+                }
                 if sender_closed {
                     close.close(super::super::connection::CloseReason::ProtocolClosed);
-+
+                }
                 drop(receive_endpoint);
                 let metadata =
                     captured_metadata.unwrap_or_else(|| connection.info().h2_metadata(close));
                 Err(
                     super::super::super::downcast_error(Box::new(error.into_error()))
                         .with_connection(metadata),
-+
+                )
-+
+            }
-+
+        }
-+
+    }
-+
+}
-+
 /// Restores the absolute URI and removes only a pool-synthesized H1 `Host`.
 fn prepare_h2_request(request: &mut Request<SdkBody>, absolute_uri: &http_1x::Uri) {
     *request.uri_mut() = absolute_uri.clone();
     if request
         .extensions_mut()
         .remove::<H1HostHeaderInserted>()
         .is_some()
-+
+    {
         request.headers_mut().remove(http_1x::header::HOST);
-+
+    }
-+
+}
-+
 /// 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,
-+
+}
-+
 /// Applies retry authority to a request that Hyper did not accept.
 ///
 /// Pool-side checks always reacquire because no protocol code observed the
 /// request. A returned Hyper envelope reacquires only after prior successful
 /// use proves that replacing a stale pooled generation is appropriate.
 #[allow(
     clippy::result_large_err,
     reason = "ConnectorError preserves SDK classification and connection metadata"
 )]
 fn finish_unaccepted_request(
     request: Request<SdkBody>,
     reused: bool,
     stage: UnacceptedStage,
     error: ConnectorError,
 ) -> Result<H2DispatchResult, ConnectorError> {
     if stage == UnacceptedStage::BeforeHyper || reused {
         Ok(H2DispatchResult::Reacquire(Box::new(H2Reacquisition {
             request,
             error,
         })))
     } else {
         Err(error)
-+
+    }
-+
+}
-+
 /// A fresh HTTP/2 generation closed before accepting its first request.
 #[derive(Debug)]
 struct H2ConnectionClosedBeforeDispatch;
-+
 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")
-+
+    }
-+
+}
-+
 impl std::error::Error for H2ConnectionClosedBeforeDispatch {}
-+
 /// Handle retained in a request extension after its body is wrapped once.
 #[derive(Clone)]
 struct H2RequestBodyHandle {
     /// Re-armable send endpoint shared with the wrapped request body.
     slot: Arc<Mutex<Option<H2LeaseEndpoint>>>,
-+
+}
-+
 impl H2RequestBodyHandle {
     /// Wraps an unwrapped body and arms the current send endpoint.
     fn arm(request: &mut Request<SdkBody>, endpoint: H2LeaseEndpoint) -> Self {
         let is_end_stream = request.body().is_end_stream();
         let handle = request
             .extensions()
             .get::<Self>()
             .cloned()
             .unwrap_or_else(|| {
                 let handle = Self {
                     slot: Arc::new(Mutex::new(None)),
                 };
                 let body = std::mem::replace(request.body_mut(), SdkBody::taken());
                 *request.body_mut() = SdkBody::from_body_1_x(H2RequestBody {
                     inner: body,
                     slot: handle.slot.clone(),
                 });
                 request.extensions_mut().insert(handle.clone());
                 handle
             });
         arm_send_endpoint(&handle.slot, endpoint);
         if is_end_stream {
             finish_send_endpoint(&handle.slot);
-+
+        }
         handle
-+
+    }
-+
     /// Disarms an endpoint after Hyper returns the request unaccepted.
     fn clear(&self) {
         finish_send_endpoint(&self.slot);
-+
+    }
-+
+}
-+
 /// Request body whose endpoint can be re-armed after certified non-acceptance.
 struct H2RequestBody {
     /// Original SDK body wrapped exactly once.
     inner: SdkBody,
     /// Endpoint replaced when Hyper returns an unaccepted request.
     slot: Arc<Mutex<Option<H2LeaseEndpoint>>>,
-+
+}
-+
 impl Body for H2RequestBody {
     type Data = <SdkBody as Body>::Data;
     type Error = <SdkBody as Body>::Error;
-+
     fn poll_frame(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
         let result = Pin::new(&mut self.inner).poll_frame(cx);
         if matches!(result, Poll::Ready(None) | Poll::Ready(Some(Err(_)))) {
             finish_send_endpoint(&self.slot);
-+
+        }
         result
-+
+    }
-+
     fn is_end_stream(&self) -> bool {
         self.inner.is_end_stream()
-+
+    }
-+
     fn size_hint(&self) -> hyper::body::SizeHint {
         self.inner.size_hint()
-+
+    }
-+
+}
-+
 impl Drop for H2RequestBody {
     fn drop(&mut self) {
         finish_send_endpoint(&self.slot);
-+
+    }
-+
+}
-+
 /// Response body that owns the receive endpoint through stream completion.
 struct H2ResponseBody {
     /// Hyper response stream.
     inner: hyper::body::Incoming,
     /// Receive endpoint completed on terminal frame, error, or drop.
     receive: Option<H2LeaseEndpoint>,
-+
+}
-+
 impl H2ResponseBody {
     /// Wraps a response and completes an endpoint already at end stream.
     fn new(inner: hyper::body::Incoming, receive: H2LeaseEndpoint) -> Self {
         let is_end_stream = inner.is_end_stream();
         Self {
             inner,
             receive: retain_receive_endpoint(is_end_stream, receive),
-+
+        }
-+
+    }
-+
+}
-+
 /// Retains a receive endpoint only while response frames may remain.
 fn retain_receive_endpoint(
     is_end_stream: bool,
     receive: H2LeaseEndpoint,
 ) -> Option<H2LeaseEndpoint> {
     if is_end_stream {
         receive.complete();
         None
     } else {
         Some(receive)
-+
+    }
-+
+}
-+
 impl Body for H2ResponseBody {
     type Data = <hyper::body::Incoming as Body>::Data;
     type Error = <hyper::body::Incoming as Body>::Error;
-+
     fn poll_frame(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
         let result = Pin::new(&mut self.inner).poll_frame(cx);
         if matches!(result, Poll::Ready(None) | Poll::Ready(Some(Err(_)))) {
             if let Some(endpoint) = self.receive.take() {
                 endpoint.complete();
-+
+            }
-+
+        }
         result
-+
+    }
-+
     fn is_end_stream(&self) -> bool {
         self.inner.is_end_stream()
-+
+    }
-+
     fn size_hint(&self) -> hyper::body::SizeHint {
         self.inner.size_hint()
-+
+    }
-+
+}
-+
 fn arm_send_endpoint(slot: &Arc<Mutex<Option<H2LeaseEndpoint>>>, endpoint: H2LeaseEndpoint) {
     let previous = slot.lock().replace(endpoint);
     drop(previous);
-+
+}
-+
 /// Terminates the send endpoint after upload completion, error, or rejection.
 fn finish_send_endpoint(slot: &Arc<Mutex<Option<H2LeaseEndpoint>>>) {
     let endpoint = slot.lock().take();
     if let Some(endpoint) = endpoint {
         endpoint.complete();
-+
+    }
-+
+}
-+
 #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
 mod tests {
     use super::*;
     use crate::client::pool::cell::OriginCell;
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::{EligibilityGroup, PartitionId};
     use crate::sync::Arc;
     use http_1x::uri::Scheme;
     use http_body_util::BodyExt;
-+
     /// Request body that fails on its first frame.
     struct FailingBody;
-+
     impl Body for FailingBody {
         type Data = hyper::body::Bytes;
         type Error = std::io::Error;
-+
         fn poll_frame(
             self: Pin<&mut Self>,
             _cx: &mut Context<'_>,
         ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
             Poll::Ready(Some(Err(std::io::Error::other("request body failed"))))
-+
+        }
-+
+    }
-+
     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,
         ))
-+
+    }
-+
     #[tokio::test]
     async fn returned_request_rearms_one_body_wrapper() {
         let cell = cell();
         let mut request = Request::post("https://example.com/upload")
             .body(SdkBody::from("payload"))
             .unwrap();
-+
         let (first_endpoint, first_probe) = H2LeaseEndpoint::send_for_test(&cell);
         let first = H2RequestBodyHandle::arm(&mut request, first_endpoint);
         let first_slot = first.slot.clone();
         assert!(!first_probe.send_complete());
-+
         let (second_endpoint, second_probe) = H2LeaseEndpoint::send_for_test(&cell);
         let second = H2RequestBodyHandle::arm(&mut request, second_endpoint);
         assert!(
             Arc::ptr_eq(&first_slot, &second.slot),
             "rearming replaced the request-body wrapper"
         );
         assert!(
             first_probe.send_complete(),
             "rearming did not cancel the prior prospective endpoint"
         );
         assert!(!second_probe.send_complete());
-+
         second.clear();
         assert!(
             second_probe.send_complete(),
             "returned request retained its rejected send endpoint"
         );
         let body = request
             .into_body()
             .collect()
             .await
             .expect("wrapped request body failed")
             .to_bytes();
         assert_eq!(hyper::body::Bytes::from_static(b"payload"), body);
-+
+    }
-+
     #[test]
     fn empty_request_body_completes_its_send_endpoint_when_armed() {
         let cell = cell();
         let mut request = Request::new(SdkBody::empty());
         let (endpoint, probe) = H2LeaseEndpoint::send_for_test(&cell);
-+
         H2RequestBodyHandle::arm(&mut request, endpoint);
-+
         assert!(probe.send_complete());
-+
+    }
     #[test]
     fn dropping_request_body_completes_its_send_endpoint() {
         let cell = cell();
         let mut request = Request::new(SdkBody::from("payload"));
         let (endpoint, probe) = H2LeaseEndpoint::send_for_test(&cell);
-+
         H2RequestBodyHandle::arm(&mut request, endpoint);
         let body = std::mem::replace(request.body_mut(), SdkBody::empty());
         drop(body);
-+
         assert!(probe.send_complete());
-+
+    }
-+
     #[tokio::test]
     async fn request_body_error_completes_its_send_endpoint() {
         let cell = cell();
         let mut request = Request::new(SdkBody::from_body_1_x(FailingBody));
         let (endpoint, probe) = H2LeaseEndpoint::send_for_test(&cell);
-+
         H2RequestBodyHandle::arm(&mut request, endpoint);
         let frame = request
             .body_mut()
             .frame()
             .await
             .expect("failing body omitted its error frame");
-+
         assert!(frame.is_err());
         assert!(probe.send_complete());
-+
+    }
-+
     #[test]
     fn bodyless_response_completes_its_receive_endpoint_without_drop() {
         let cell = cell();
         let (endpoint, probe) = H2LeaseEndpoint::receive_for_test(&cell);
-+
         let retained = retain_receive_endpoint(true, endpoint);
-+
         assert!(retained.is_none());
         assert!(probe.receive_complete());
-+
+    }
-+
     #[test]
     fn h2_preparation_restores_uri_and_removes_only_synthesized_host() {
         let absolute: http_1x::Uri = "https://example.com/resource".parse().unwrap();
         let mut request = Request::get("/resource").body(SdkBody::empty()).unwrap();
         request
             .headers_mut()
             .insert(http_1x::header::HOST, "example.com".parse().unwrap());
         request.extensions_mut().insert(H1HostHeaderInserted);
-+
         prepare_h2_request(&mut request, &absolute);
-+
         assert_eq!(&absolute, request.uri());
         assert!(!request.headers().contains_key(http_1x::header::HOST));
         assert!(request.extensions().get::<H1HostHeaderInserted>().is_none());
-+
         let mut user_host = Request::get("/resource").body(SdkBody::empty()).unwrap();
         user_host
             .headers_mut()
             .insert(http_1x::header::HOST, "signed.example".parse().unwrap());
         prepare_h2_request(&mut user_host, &absolute);
         assert_eq!("signed.example", user_host.headers()[http_1x::header::HOST]);
-+
+    }
-+
     #[test]
     fn retry_authority_distinguishes_pool_checks_from_hyper_returns() {
         let checked_before_hyper = finish_unaccepted_request(
             Request::new(SdkBody::empty()),
             false,
             UnacceptedStage::BeforeHyper,
             ConnectorError::user("pre-Hyper failure".into()),
         );
         assert!(matches!(
             checked_before_hyper,
             Ok(H2DispatchResult::Reacquire(_))
         ));
-+
         let fresh_hyper_return = finish_unaccepted_request(
             Request::new(SdkBody::empty()),
             false,
             UnacceptedStage::ReturnedByHyper,
             ConnectorError::user("fresh Hyper failure".into()),
         );
         assert!(fresh_hyper_return.is_err());
-+
         let reused_hyper_return = finish_unaccepted_request(
             Request::new(SdkBody::empty()),
             true,
             UnacceptedStage::ReturnedByHyper,
             ConnectorError::user("reused failure".into()),
         );
         assert!(matches!(
             reused_hyper_return,
             Ok(H2DispatchResult::Reacquire(_))
         ));
-+
+    }
-+
+}

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

@@ -1,1 +0,380 @@

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

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

@@ -1,1 +155,0 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! HTTP/1 connection establishment.
 //!
 //! Establishment connects a transport, performs Hyper's client handshake,
 //! installs the resulting exclusive sender in the origin cell, and submits
 //! the protocol driver to the connection-owning partition.
 //! Establishment receives an already connected transport, performs Hyper's
 //! client handshake, installs the resulting exclusive sender in the origin
 //! cell, and submits the protocol driver to the connection-owning partition.
-11
 use super::super::cell::h1::{H1CloseHandle, H1DriverGuard, H1Selection, H1Sender};
 use super::super::cell::{EstablishmentPermit, OriginCell};
 use super::super::connection::{
     CloseReason, ConnectionInfo, ConnectionIo, ConnectionState, NegotiatedProtocol,
 };
 use super::super::dispatch::AcquisitionContext;
 use super::next_connection_id;
 use crate::client::connect::BoxConn;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use std::error::Error;
 use std::fmt;
 use hyper_util::client::legacy::connect::Connected;
-23
 /// Connects, handshakes, installs, and starts the owner-partition driver.
 /// Handshakes, installs, and starts the owner-partition driver.
 ///
 /// `permit` remains the sole owner of optional bounded capacity until the
 /// connection state is created. An earlier failure returns that capacity
 /// through the permit's fallback. A handshake failure logically closes the
 /// new connection state, while success transfers driver ownership to the
 /// partition runtime and returns the installed sender selection.
 pub(in crate::client::pool) async fn establish_h1(
     context: AcquisitionContext,
     permit: EstablishmentPermit,
     io: BoxConn,
     connected: Connected,
 ) -> Result<H1Selection, ConnectorError> {
     let request_partition = context.partition.id();
     let connection_cell = context.cell.clone();
     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(),
         "HTTP/1 connection establishment started"
     );
-47
     let result = connect_handshake_and_install_h1(context, permit).await;
     let result = handshake_and_install_h1(context, permit, io, connected).await;
     match &result {
         Ok(selection) => {
             let connection = selection.connection();
             tracing::debug!(
                 connection_id = %connection.id(),
                 request_partition = ?request_partition,
                 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 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
 /// Connects, handshakes, and installs one HTTP/1 connection.
 async fn connect_handshake_and_install_h1(
 /// Handshakes and installs one already connected HTTP/1 transport.
 async fn handshake_and_install_h1(
     context: AcquisitionContext,
     permit: EstablishmentPermit,
     io: BoxConn,
     connected: Connected,
 ) -> Result<H1Selection, ConnectorError> {
     let AcquisitionContext {
         pool,
         partition,
         cell,
         absolute_uri,
         absolute_uri: _,
         owner_spawner,
         connect_timeout,
         connect_timeout: _,
     } = context;
-90
     let io = pool
         .transport
         .connect(&partition, absolute_uri, connect_timeout)
         .await
         .map_err(super::super::super::downcast_error)?;
     let connected = io.connected();
     if connected.is_negotiated_h2() {
         return Err(ConnectorError::user(UnsupportedNegotiatedProtocol.into()));
--
+    }
--
     let id =
         next_connection_id(&pool).map_err(|error| ConnectorError::other(error.into(), None))?;
     let info = ConnectionInfo::new(
         id,
         cell.id().origin().clone(),
         partition.id(),
         NegotiatedProtocol::Http1,
         connected,
     );
     let (connection, physical) = match permit.into_lease() {
         Some(lease) => ConnectionState::bounded(info, lease),
         None => ConnectionState::unbounded(info),
     };
     let (connection, physical) = ConnectionState::pending_open(info);
     let io = ConnectionIo::new(io, physical);
-102
     let (sender, driver) = match hyper::client::conn::http1::Builder::new()
         .handshake::<_, SdkBody>(io)
         .await
-106
+    {
         Ok(established) => established,
         Err(error) => {
             connection.logical_close(CloseReason::ProtocolClosed);
             return Err(super::super::super::downcast_error(Box::new(error)));
-111
+        }
     };
-113
     if let Err(lease) = connection.open(permit.into_lease()) {
         drop(lease);
         connection.logical_close(CloseReason::ProtocolClosed);
         return Err(ConnectorError::io(
             "HTTP/1 connection closed before installation".into(),
         ));
-+
+    }
-+
     let selection =
         OriginCell::install_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 {
         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"
             );
-138
+        }
         driver_guard.protocol_closed();
     }));
     Ok(selection)
-142
+}
--
 /// The cleartext HTTP/1 path rejected a connector-selected HTTP/2 transport.
 #[derive(Debug)]
 struct UnsupportedNegotiatedProtocol;
--
 impl fmt::Display for UnsupportedNegotiatedProtocol {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str("the HTTP/1 pool path cannot install an HTTP/2 transport")
--
+    }
--
+}
--
 impl Error for UnsupportedNegotiatedProtocol {}

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

@@ -0,1 +0,475 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-+
 //! HTTP/2 flight convergence, handshake, and driver installation.
 //!
 //! ALPN-selected attempts first enter one cell-local transition: use an
 //! accepting generation, join the current flight, or become the flight's
 //! owner task. Only the owner performs Hyper's HTTP/2 handshake. It opens the
 //! connection, installs the generation, submits the driver on the partition
 //! spawner, and then disarms flight cleanup. Participants own only their
 //! waiter entries; losing attempts close their unused transport and return
 //! their capacity before this function reports transfer. The connector timeout
 //! ends before this phase; no separate pool timeout covers the HTTP/2
 //! handshake or driver submission.
-+
 use super::super::cell::h2::{
     H2CloseHandle, H2DriverGuard, H2FlightId, H2FlightInstall, H2GenerationJoin, H2Sender,
 };
 use super::super::cell::{AcquisitionResult, EstablishmentPermit, OriginCell, WaiterId};
 use super::super::connection::{
     CloseReason, ConnectionInfo, ConnectionIo, ConnectionState, NegotiatedProtocol,
 };
 use super::super::dispatch::AcquisitionContext;
 use super::super::partition::DriverSpawner;
 use super::{next_connection_id, EstablishmentOutcome};
 use crate::client::connect::BoxConn;
 use aws_smithy_runtime_api::box_error::BoxError;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use aws_smithy_types::retry::ErrorKind;
 use hyper::rt::Executor;
 use hyper_util::client::legacy::connect::Connected;
 use std::future::Future;
 use std::sync::Arc as StdArc;
-+
 /// Hyper executor that keeps spawned HTTP/2 work on the connection partition.
 #[derive(Clone, Debug)]
 struct PartitionExecutor {
     /// Runtime placement inherited from the connection-owning partition.
     spawner: StdArc<dyn DriverSpawner>,
-+
+}
-+
 impl<F> Executor<F> for PartitionExecutor
 where
     F: Future<Output = ()> + Send + 'static,
-+
+{
     fn execute(&self, future: F) {
         self.spawner.spawn(Box::pin(future));
-+
+    }
-+
+}
-+
 /// Settles a flight if its owner-runtime task is discarded.
 struct FlightCompletionGuard {
     /// Cell that owns the exact flight.
     cell: crate::sync::Arc<OriginCell>,
     /// Flight identity protected from stale task completion.
     flight: H2FlightId,
     /// Whether drop must fail remaining participants.
     active: bool,
-+
+}
-+
 impl FlightCompletionGuard {
     fn new(cell: crate::sync::Arc<OriginCell>, flight: H2FlightId) -> Self {
         Self {
             cell,
             flight,
             active: true,
-+
+        }
-+
+    }
-+
     fn disarm(&mut self) {
         self.active = false;
-+
+    }
-+
     fn fail(&mut self, error: ConnectorError) {
         self.active = false;
         fail_participants(&self.cell, self.flight, error);
-+
+    }
-+
+}
-+
 impl Drop for FlightCompletionGuard {
     fn drop(&mut self) {
         if self.active {
             fail_participants(
                 &self.cell,
                 self.flight,
                 ConnectorError::io(Box::new(std::io::Error::other(
                     "HTTP/2 establishment task was dropped",
                 ))),
             );
-+
+        }
-+
+    }
-+
+}
-+
 /// Converges after ALPN and performs at most one HTTP/2 handshake per cell.
 pub(super) async fn establish_h2(
     context: AcquisitionContext,
     waiter: WaiterId,
     permit: EstablishmentPermit,
     io: BoxConn,
     connected: Connected,
 ) -> EstablishmentOutcome {
     loop {
         match context.cell.install_or_join_h2_flight(waiter) {
             H2FlightInstall::Accepting(generation) => {
                 tracing::trace!(
                     request_partition = ?context.partition.id(),
                     connection_partition = ?context.cell.id().partition(),
                     origin_scheme = %context.cell.id().origin().scheme(),
                     origin_host = context.cell.id().origin().host(),
                     origin_port = ?context.cell.id().origin().port(),
                     h2_generation = ?generation,
                     "HTTP/2 establishment found an accepting generation"
                 );
                 match OriginCell::join_h2_generation(&context.cell, waiter, generation) {
                     H2GenerationJoin::GenerationChanged => continue,
                     H2GenerationJoin::Joined | H2GenerationJoin::WaiterCompleted => {
                         drop(io);
                         drop(permit);
                         return EstablishmentOutcome::Transferred;
-+
+                    }
-+
+                }
-+
+            }
             H2FlightInstall::Joined => {
                 tracing::trace!(
                     request_partition = ?context.partition.id(),
                     connection_partition = ?context.cell.id().partition(),
                     origin_scheme = %context.cell.id().origin().scheme(),
                     origin_host = context.cell.id().origin().host(),
                     origin_port = ?context.cell.id().origin().port(),
                     "HTTP/2 establishment joined the active flight"
                 );
                 drop(io);
                 drop(permit);
                 return EstablishmentOutcome::Transferred;
-+
+            }
             H2FlightInstall::WaiterCompleted => {
                 tracing::trace!(
                     request_partition = ?context.partition.id(),
                     connection_partition = ?context.cell.id().partition(),
                     origin_scheme = %context.cell.id().origin().scheme(),
                     origin_host = context.cell.id().origin().host(),
                     origin_port = ?context.cell.id().origin().port(),
                     "HTTP/2 establishment waiter already completed"
                 );
                 drop(io);
                 drop(permit);
                 return EstablishmentOutcome::Transferred;
-+
+            }
             H2FlightInstall::Driver(flight) => {
                 tracing::trace!(
                     request_partition = ?context.partition.id(),
                     connection_partition = ?context.cell.id().partition(),
                     origin_scheme = %context.cell.id().origin().scheme(),
                     origin_host = context.cell.id().origin().host(),
                     origin_port = ?context.cell.id().origin().port(),
                     h2_flight = ?flight,
                     "HTTP/2 establishment started a flight"
                 );
                 drive_flight(context, flight, permit, io, connected).await;
                 return EstablishmentOutcome::Transferred;
-+
+            }
-+
+        }
-+
+    }
-+
+}
-+
 /// Handshakes, installs, and publishes one winning flight.
 async fn drive_flight(
     context: AcquisitionContext,
     flight: H2FlightId,
     permit: EstablishmentPermit,
     io: BoxConn,
     connected: Connected,
 ) {
     let mut completion = FlightCompletionGuard::new(context.cell.clone(), flight);
     let id = match next_connection_id(&context.pool) {
         Ok(id) => id,
         Err(error) => {
             completion.fail(ConnectorError::other(Box::new(error), None));
             return;
-+
+        }
     };
     let info = ConnectionInfo::new(
         id,
         context.cell.id().origin().clone(),
         context.partition.id(),
         NegotiatedProtocol::Http2,
         connected,
     );
     let (connection, physical) = ConnectionState::pending_open(info);
     let io = ConnectionIo::new(io, physical);
     let executor = PartitionExecutor {
         spawner: context.owner_spawner.clone(),
     };
     let (sender, driver) = match hyper::client::conn::http2::Builder::new(executor)
         .handshake::<_, SdkBody>(io)
         .await
-+
+    {
         Ok(established) => established,
         Err(error) => {
             connection.logical_close(CloseReason::ProtocolClosed);
             completion.fail(super::super::super::downcast_error(Box::new(error)));
             return;
-+
+        }
     };
-+
     if let Err(lease) = connection.open(permit.into_lease()) {
         drop(lease);
         connection.logical_close(CloseReason::ProtocolClosed);
         completion.fail(ConnectorError::io(Box::new(std::io::Error::other(
             "HTTP/2 connection closed before installation",
         ))));
         return;
-+
+    }
-+
     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);
             completion.fail(ConnectorError::io(Box::new(std::io::Error::other(
                 "HTTP/2 flight became stale before installation",
             ))));
             return;
-+
+        }
     };
-+
     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 {
         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"
             );
-+
+        }
         driver_guard.protocol_closed();
     }));
-+
     completion.disarm();
     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"
     );
-+
+}
-+
 /// Connector classification copied to each participant in one failed flight.
 #[derive(Clone, Copy, Debug)]
 enum SharedFailureKind {
     /// Connector timeout classification.
     Timeout,
     /// Caller or configuration error classification.
     User,
     /// Retryable transport I/O classification.
     Io,
     /// Other connector classification and optional retry kind.
     Other(Option<ErrorKind>),
-+
+}
-+
 /// Cloneable wrapper that preserves one flight failure for every participant.
 #[derive(Clone, Debug)]
 struct SharedFlightFailure {
     /// Original source retained for every participant result.
     source: StdArc<BoxError>,
     /// Connector classification copied without consuming the source.
     kind: SharedFailureKind,
-+
+}
-+
 impl SharedFlightFailure {
     fn new(error: ConnectorError) -> Self {
         let kind = if error.is_timeout() {
             SharedFailureKind::Timeout
         } else if error.is_user() {
             SharedFailureKind::User
         } else if error.is_io() {
             SharedFailureKind::Io
         } else {
             debug_assert!(error.is_other());
             SharedFailureKind::Other(error.as_other())
         };
         Self {
             source: StdArc::new(error.into_source()),
             kind,
-+
+        }
-+
+    }
-+
     fn connector_error(&self) -> ConnectorError {
         let source: BoxError = Box::new(self.clone());
         match self.kind {
             SharedFailureKind::Timeout => ConnectorError::timeout(source),
             SharedFailureKind::User => ConnectorError::user(source),
             SharedFailureKind::Io => ConnectorError::io(source),
             SharedFailureKind::Other(kind) => ConnectorError::other(source, kind),
-+
+        }
-+
+    }
-+
+}
-+
 impl std::fmt::Display for SharedFlightFailure {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         std::fmt::Display::fmt(&self.source, f)
-+
+    }
-+
+}
-+
 impl std::error::Error for SharedFlightFailure {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         Some(self.source.as_ref().as_ref())
-+
+    }
-+
+}
-+
 /// Fails every participant still retained by one exact flight.
 fn fail_participants(cell: &OriginCell, flight: H2FlightId, error: ConnectorError) {
     let Some(participants) = cell.fail_h2_flight(flight) else {
         return;
     };
     let error = SharedFlightFailure::new(error);
     for participant in participants {
         cell.complete_establishment(
             participant,
             AcquisitionResult::Failed(error.connector_error()),
         );
-+
+    }
-+
+}
-+
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
     use crate::client::pool::admission::ProtocolRequirement;
     use crate::client::pool::cell::AcquisitionEvent;
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::EligibilityGroup;
     use http_1x::uri::Scheme;
     use std::error::Error as _;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::task::{Context, Poll, Waker};
-+
     fn cell() -> crate::sync::Arc<OriginCell> {
         crate::sync::Arc::new(OriginCell::new(
             super::super::super::partition::PartitionId::from_index(1),
             OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
             EligibilityGroup::Pool,
             None,
             None,
         ))
-+
+    }
-+
     fn launching_waiter(cell: &crate::sync::Arc<OriginCell>) -> WaiterId {
         let waiter = OriginCell::register_waiter(cell, ProtocolRequirement::H2Required);
         let event = cell.poll_waiter(waiter, &mut Context::from_waker(Waker::noop()));
         let Poll::Ready(AcquisitionEvent::Establish(permit)) = event else {
             panic!("new H2 waiter did not receive establishment authority");
         };
         assert!(cell.start_establishment(waiter));
         drop(permit);
         waiter
-+
+    }
-+
     fn failed_event(cell: &OriginCell, waiter: WaiterId) -> ConnectorError {
         let event = cell.poll_waiter(waiter, &mut Context::from_waker(Waker::noop()));
         let Poll::Ready(AcquisitionEvent::Complete(AcquisitionResult::Failed(error))) = event
         else {
             panic!("flight participant did not receive a failure");
         };
         error
-+
+    }
-+
     #[test]
     fn flight_failure_reaches_every_live_participant() {
         let cell = cell();
         let first = launching_waiter(&cell);
         let second = launching_waiter(&cell);
         let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(first) else {
             panic!("first participant did not become the flight driver");
         };
         assert!(matches!(
             cell.install_or_join_h2_flight(second),
             H2FlightInstall::Joined
         ));
-+
         let mut completion = FlightCompletionGuard::new(cell.clone(), flight);
         completion.fail(ConnectorError::io(Box::new(std::io::Error::other(
             "synthetic HTTP/2 handshake failure",
         ))));
-+
         for waiter in [first, second] {
             let error = failed_event(&cell, waiter);
             assert!(error.is_io());
             let source = error.source().expect("flight failure lost its source");
             assert_eq!("synthetic HTTP/2 handshake failure", source.to_string());
             assert!(
                 source
                     .source()
                     .expect("flight failure lost its original source")
                     .downcast_ref::<std::io::Error>()
                     .is_some(),
                 "flight failure did not preserve the original error"
             );
-+
+        }
-+
+    }
-+
     #[test]
     fn task_drop_fails_only_participants_still_owned_by_the_flight() {
         let cell = cell();
         let live = launching_waiter(&cell);
         let cancelled = launching_waiter(&cell);
         let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(live) else {
             panic!("first participant did not become the flight driver");
         };
         assert!(matches!(
             cell.install_or_join_h2_flight(cancelled),
             H2FlightInstall::Joined
         ));
         assert!(OriginCell::cancel_waiter(&cell, cancelled));
-+
         drop(FlightCompletionGuard::new(cell.clone(), flight));
-+
         let error = failed_event(&cell, live);
         assert!(error.is_io());
         assert_eq!(
             "HTTP/2 establishment task was dropped",
             error
                 .source()
                 .expect("task-drop failure lost its source")
                 .to_string()
         );
         let successor = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
         assert_ne!(cancelled, successor);
         assert!(OriginCell::cancel_waiter(&cell, successor));
-+
+    }
-+
     #[derive(Debug)]
     struct CountingSpawner {
         submissions: StdArc<AtomicUsize>,
-+
+    }
-+
     impl DriverSpawner for CountingSpawner {
         fn spawn(&self, future: std::pin::Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             self.submissions.fetch_add(1, Ordering::Relaxed);
             drop(future);
-+
+        }
-+
+    }
-+
     #[test]
     fn hyper_executor_submits_work_to_the_partition_spawner() {
         let submissions = StdArc::new(AtomicUsize::new(0));
         let executor = PartitionExecutor {
             spawner: StdArc::new(CountingSpawner {
                 submissions: submissions.clone(),
             }),
         };
-+
         executor.execute(async {});
-+
         assert_eq!(1, submissions.load(Ordering::Relaxed));
-+
+    }
-+
+}

Pages: 1 2 3 4 5 6 7