AWS SDK

rev. 174400987dccd7e137fefa96b1143d21c7ddfb78

Files changed:

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

@@ -0,1 +0,716 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-+
 //! Connection pool builder.
 //!
 //! Entry point: [`SharedPool::builder`](super::SharedPool::builder).
-+
 use std::sync::Arc;
 use std::time::Duration;
-+
 use aws_smithy_runtime_api::client::dns::{ResolveDns, SharedDnsResolver};
 use aws_smithy_runtime_api::shared::IntoShared;
 use hyper_util::client::legacy::connect::dns::Name as DnsName;
 use hyper_util::client::legacy::connect::HttpConnector as HyperHttpConnector;
 use hyper_util::client::proxy::matcher::Matcher as ProxyMatcher;
-+
 use super::connection::ConnectionEventListener;
 use super::partition::{CrossPartitionPolicy, Partition};
 use super::{BoxError, ConnectionPool, PoolConfig};
 use crate::client::dns::HyperUtilResolver;
 use crate::client::proxy::ProxyConfig;
 use crate::client::tls;
 use crate::client::{TlsProviderSelected, TlsUnset};
 use crate::tls::TlsContext;
-+
 /// Default idle-connection eviction timeout, applied when the caller does
 /// not configure one.
 const DEFAULT_POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(60);
-+
 /// Builder for a [`SharedPool`].
 ///
 /// Configures pool-wide settings: TLS, DNS, connection limits, idle
 /// eviction, proxy, partition topology, and connection event listening.
 ///
 /// Type-state ensures TLS is configured before [`build_https`] is
 /// callable; calling [`tls_provider`] transitions the builder into the
 /// state where [`build_https`] is available.
 ///
 /// [`build_https`]: Builder::build_https
 /// [`tls_provider`]: Builder::tls_provider
 /// [`SharedPool`]: super::SharedPool
 #[derive(Clone)]
 pub struct Builder<Tls = TlsUnset> {
     pool_idle_timeout: Option<Option<Duration>>,
     tcp_nodelay: bool,
     tcp_keepalive: Option<Option<Duration>>,
     max_connections: Option<usize>,
     max_connections_per_host: Option<usize>,
     proxy_config: Option<ProxyConfig>,
     connection_event_listener: Option<Arc<dyn ConnectionEventListener>>,
     cross_partition_policy: CrossPartitionPolicy,
     dns_resolver: Option<SharedDnsResolver>,
     partitions: Vec<Partition>,
     tls: Tls,
-+
+}
-+
 impl<Tls: std::fmt::Debug> std::fmt::Debug for Builder<Tls> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Builder")
             .field("pool_idle_timeout", &self.pool_idle_timeout)
             .field("tcp_nodelay", &self.tcp_nodelay)
             .field("tcp_keepalive", &self.tcp_keepalive)
             .field("max_connections", &self.max_connections)
             .field("max_connections_per_host", &self.max_connections_per_host)
             .field("proxy_config", &self.proxy_config)
             .field(
                 "connection_event_listener",
                 &self.connection_event_listener.as_ref().map(|_| ".."),
-+
+            )
             .field("cross_partition_policy", &self.cross_partition_policy)
             .field("dns_resolver", &self.dns_resolver.as_ref().map(|_| ".."))
             .field("partitions", &self.partitions.len())
             .finish()
-+
+    }
-+
+}
-+
 impl Default for Builder<TlsUnset> {
     fn default() -> Self {
         Self {
             pool_idle_timeout: None,
             tcp_nodelay: true,
             tcp_keepalive: None,
             max_connections: None,
             max_connections_per_host: None,
             proxy_config: None,
             connection_event_listener: None,
             cross_partition_policy: CrossPartitionPolicy::default(),
             dns_resolver: None,
             partitions: Vec::new(),
             tls: TlsUnset {},
-+
+        }
-+
+    }
-+
+}
-+
 // Methods available in any TLS state.
 impl<Tls> Builder<Tls> {
     /// Set the pool idle timeout.
     ///
     /// Connections idle longer than this duration are evicted from the
     /// pool. Set below the server's idle timeout to avoid dispatching on a
     /// connection the server has already closed.
     ///
     /// Unset, the pool uses a default of 60 seconds. Pass `Some(duration)`
     /// to override, or `None` to disable idle eviction entirely.
     pub fn pool_idle_timeout<D>(mut self, timeout: D) -> Self
     where
         D: Into<Option<Duration>>,
-+
+    {
         self.pool_idle_timeout = Some(timeout.into());
         self
-+
+    }
-+
     /// This is the mutable version of [`pool_idle_timeout`](Self::pool_idle_timeout).
     ///
     /// The outer `None` selects the default; `Some(None)` disables idle
     /// eviction; `Some(Some(d))` sets the timeout to `d`.
     pub fn set_pool_idle_timeout(&mut self, timeout: Option<Option<Duration>>) -> &mut Self {
         self.pool_idle_timeout = timeout;
         self
-+
+    }
-+
     /// Set TCP_NODELAY on connections. Default: `true`.
     pub fn tcp_nodelay(mut self, nodelay: bool) -> Self {
         self.tcp_nodelay = nodelay;
         self
-+
+    }
-+
     /// This is the mutable version of [`tcp_nodelay`](Self::tcp_nodelay).
     pub fn set_tcp_nodelay(&mut self, nodelay: bool) -> &mut Self {
         self.tcp_nodelay = nodelay;
         self
-+
+    }
-+
     /// Set the TCP keepalive idle time.
     ///
     /// Enables `SO_KEEPALIVE` with the given idle time before the first
     /// probe. Keepalive detects dead peers faster than idle eviction
     /// alone, notably for long-lived H2 connections.
     ///
     /// Keepalive is disabled by default. Pass `Some(duration)` to enable
     /// it with that idle time; `None` leaves it disabled.
     pub fn tcp_keepalive<D>(mut self, time: D) -> Self
     where
         D: Into<Option<Duration>>,
-+
+    {
         self.tcp_keepalive = Some(time.into());
         self
-+
+    }
-+
     /// This is the mutable version of [`tcp_keepalive`](Self::tcp_keepalive).
     ///
     /// The outer `None` selects the default; `Some(None)` disables
     /// keepalive; `Some(Some(d))` sets the idle time to `d`.
     pub fn set_tcp_keepalive(&mut self, time: Option<Option<Duration>>) -> &mut Self {
         self.tcp_keepalive = time;
         self
-+
+    }
-+
     /// Set the global maximum number of concurrent connections.
     ///
     /// Caps the total live connections in the pool across all hosts,
     /// including idle (cached) connections. New connection attempts wait
     /// when the pool is at capacity; existing connections must be evicted
     /// or closed before another can be created.
     ///
     /// Should be at least [`max_connections_per_host`](Self::max_connections_per_host)
     /// when both are set; the global limit applies on top of the per-host
     /// limit.
     pub fn max_connections(mut self, n: usize) -> Self {
         self.max_connections = Some(n);
         self
-+
+    }
-+
     /// This is the mutable version of [`max_connections`](Self::max_connections).
     ///
     /// `None` leaves the global limit unset (unbounded).
     pub fn set_max_connections(&mut self, n: Option<usize>) -> &mut Self {
         self.max_connections = n;
         self
-+
+    }
-+
     /// Set the maximum number of concurrent connections per host.
     ///
     /// Each unique (scheme, authority) pair has an independent connection
     /// budget. This limit is independent of `max_connections`; both can
     /// be set and are enforced simultaneously.
     pub fn max_connections_per_host(mut self, n: usize) -> Self {
         self.max_connections_per_host = Some(n);
         self
-+
+    }
-+
     /// This is the mutable version of
     /// [`max_connections_per_host`](Self::max_connections_per_host).
     ///
     /// `None` leaves the per-host limit unset (unbounded).
     pub fn set_max_connections_per_host(&mut self, n: Option<usize>) -> &mut Self {
         self.max_connections_per_host = n;
         self
-+
+    }
-+
     /// Route connections through an HTTP/HTTPS/SOCKS proxy.
     ///
     /// Per-host proxy resolution is stable for the lifetime of the
     /// pool: all connections to a given authority follow the same
     /// proxy decision. HTTPS through an HTTP proxy uses `CONNECT`
     /// tunneling.
     ///
     /// Pass [`ProxyConfig::from_env`] to read configuration from the
     /// `HTTP_PROXY`, `HTTPS_PROXY`, and `NO_PROXY` environment variables.
     pub fn proxy_config(mut self, config: ProxyConfig) -> Self {
         self.proxy_config = Some(config);
         self
-+
+    }
-+
     /// This is the mutable version of [`proxy_config`](Self::proxy_config).
     pub fn set_proxy_config(&mut self, config: Option<ProxyConfig>) -> &mut Self {
         self.proxy_config = config;
         self
-+
+    }
-+
     /// Set a listener for connection lifecycle events (created, reused, closed, failed).
     pub fn connection_event_listener(mut self, listener: Arc<dyn ConnectionEventListener>) -> Self {
         self.connection_event_listener = Some(listener);
         self
-+
+    }
-+
     /// This is the mutable version of [`connection_event_listener`](Self::connection_event_listener).
     pub fn set_connection_event_listener(
         &mut self,
         listener: Option<Arc<dyn ConnectionEventListener>>,
     ) -> &mut Self {
         self.connection_event_listener = listener;
         self
-+
+    }
-+
     /// Set the policy that governs checkout when the local partition has
     /// no idle connection and the pool is at capacity.
     ///
     /// Defaults to [`CrossPartitionPolicy::Never`]. Has no observable
     /// effect with a single partition or when the pool stays under
     /// capacity.
     pub fn cross_partition_policy(mut self, policy: CrossPartitionPolicy) -> Self {
         self.cross_partition_policy = policy;
         self
-+
+    }
-+
     /// This is the mutable version of [`cross_partition_policy`](Self::cross_partition_policy).
     pub fn set_cross_partition_policy(&mut self, policy: CrossPartitionPolicy) -> &mut Self {
         self.cross_partition_policy = policy;
         self
-+
+    }
-+
     /// Set a custom DNS resolver.
     ///
     /// Connections established by this pool resolve hostnames through the
     /// given resolver. Defaults to the system resolver when unset.
     pub fn dns_resolver(mut self, resolver: impl ResolveDns + 'static) -> Self {
         self.dns_resolver = Some(resolver.into_shared());
         self
-+
+    }
-+
     /// This is the mutable version of [`dns_resolver`](Self::dns_resolver).
     pub fn set_dns_resolver(&mut self, resolver: Option<SharedDnsResolver>) -> &mut Self {
         self.dns_resolver = resolver;
         self
-+
+    }
-+
     /// Declare the pool's partition topology. Each [`Partition`] carries a
     /// driver-spawner runtime and optional NIC binding. When omitted, the
     /// pool uses a single anonymous partition on the current tokio runtime.
     pub fn partitions(mut self, partitions: impl IntoIterator<Item = Partition>) -> Self {
         self.partitions = partitions.into_iter().collect();
         self
-+
+    }
-+
     /// This is the mutable version of [`partitions`](Self::partitions).
     ///
     /// `None` clears any declared topology; the pool then uses a single
     /// anonymous partition on the current tokio runtime.
     pub fn set_partitions(&mut self, partitions: Option<Vec<Partition>>) -> &mut Self {
         self.partitions = partitions.unwrap_or_default();
         self
-+
+    }
-+
+}
-+
 impl Builder<TlsUnset> {
     /// Set the TLS implementation.
     pub fn tls_provider(self, provider: tls::Provider) -> Builder<TlsProviderSelected> {
         Builder {
             pool_idle_timeout: self.pool_idle_timeout,
             tcp_nodelay: self.tcp_nodelay,
             tcp_keepalive: self.tcp_keepalive,
             max_connections: self.max_connections,
             max_connections_per_host: self.max_connections_per_host,
             proxy_config: self.proxy_config,
             connection_event_listener: self.connection_event_listener,
             cross_partition_policy: self.cross_partition_policy,
             dns_resolver: self.dns_resolver,
             partitions: self.partitions,
             tls: TlsProviderSelected {
                 provider,
                 context: TlsContext::default(),
             },
-+
+        }
-+
+    }
-+
     /// Build an HTTP client without TLS.
     #[doc(hidden)]
     pub fn build_http(mut self) -> super::SharedPool {
         let dns_resolver = self.dns_resolver.take();
         let config = PoolConfig {
             max_connections: self.max_connections,
             max_connections_per_host: self.max_connections_per_host,
             pool_idle_timeout: resolve_pool_idle_timeout(self.pool_idle_timeout),
             connection_event_listener: self.connection_event_listener.clone(),
         };
         let keepalive = resolve_tcp_keepalive(self.tcp_keepalive);
         let proxy_matcher = proxy_matcher_from(&self.proxy_config);
         let partitions = std::mem::take(&mut self.partitions);
         let policy = self.cross_partition_policy;
         let pool = match dns_resolver {
             Some(resolver) => {
                 let mut tcp = HyperHttpConnector::new_with_resolver(HyperUtilResolver { resolver });
                 tcp.set_nodelay(self.tcp_nodelay);
                 tcp.set_keepalive(keepalive);
                 build_http_pool_with_proxy(tcp, &self.proxy_config, config, partitions, policy)
-+
+            }
             None => {
                 let mut tcp = HyperHttpConnector::new();
                 tcp.set_nodelay(self.tcp_nodelay);
                 tcp.set_keepalive(keepalive);
                 build_http_pool_with_proxy(tcp, &self.proxy_config, config, partitions, policy)
-+
+            }
         };
         super::SharedPool {
             inner: Arc::new(super::SharedPoolInner {
                 pool: Arc::new(pool),
                 proxy_matcher,
             }),
-+
+        }
-+
+    }
-+
     /// Build an HTTP client from a raw TCP-level connector.
     ///
     /// The connector must be a `tower::Service<Uri>` producing an IO type
     /// that implements hyper's `Read`, `Write`, and `Connection` traits.
     /// The pool's Negotiate layer uses `Connection::connected().is_negotiated_h2()`
     /// to route connections to the H2 path.
     ///
     /// NIC binding is not applied to custom TCP connectors — the custom
     /// connector owns its own socket configuration.
     #[cfg(all(feature = "test-util", aws_sdk_unstable))]
     #[doc(hidden)]
     pub fn build_http_with_tcp_connector<C, IO>(self, connector: C) -> super::SharedPool
     where
         C: tower::Service<http_1x::Uri, Response = IO> + Clone + Send + Sync + 'static,
         C::Error: Into<BoxError> + 'static,
         C::Future: Unpin + Send + 'static,
         IO: hyper::rt::Read
             + hyper::rt::Write
             + hyper_util::client::legacy::connect::Connection
             + Unpin
             + Send
             + 'static,
-+
+    {
         let config = super::PoolConfig {
             max_connections: self.max_connections,
             max_connections_per_host: self.max_connections_per_host,
             pool_idle_timeout: resolve_pool_idle_timeout(self.pool_idle_timeout),
             connection_event_listener: self.connection_event_listener.clone(),
         };
         let policy = self.cross_partition_policy;
         let connector_factory = move |_partition: &Partition| connector.clone();
         let pool = super::build_pool(connector_factory, config, self.partitions, policy);
         super::SharedPool {
             inner: Arc::new(super::SharedPoolInner {
                 pool: Arc::new(pool),
                 proxy_matcher: None,
             }),
-+
+        }
-+
+    }
-+
+}
-+
 /// Build a no-TLS pool that honors `proxy_config`. Wraps the TCP connector
 /// with an HTTP proxy connector when configured. Emits a warning if an
 /// HTTPS proxy is set without a TLS provider; connections to such a
 /// proxy will fail at handshake time.
 fn build_http_pool_with_proxy<R>(
     tcp: HyperHttpConnector<R>,
     proxy_config: &Option<ProxyConfig>,
     config: PoolConfig,
     partitions: Vec<Partition>,
     policy: CrossPartitionPolicy,
 ) -> ConnectionPool
 where
     R: Clone + Send + Sync + 'static,
     R: tower::Service<DnsName>,
     R::Response: Iterator<Item = std::net::SocketAddr>,
     R::Future: Send,
     R::Error: Into<BoxError>,
-+
+{
     let proxy_config = proxy_config.clone().unwrap_or_else(ProxyConfig::disabled);
-+
     if proxy_config.requires_tls() {
         tracing::warn!(
             "HTTPS proxy configured but no TLS provider set. \
              Connections to HTTPS proxy servers will fail. \
              Consider configuring a TLS provider to enable TLS support."
         );
-+
+    }
-+
     if proxy_config.is_disabled() {
         let connector_factory =
             move |partition: &Partition| bind_interface(&tcp, partition.nic.as_deref());
         super::build_pool(connector_factory, config, partitions, policy)
     } else {
         let connector_factory = move |partition: &Partition| {
             crate::client::connect::HttpProxyConnector::new(
                 bind_interface(&tcp, partition.nic.as_deref()),
                 proxy_config.clone(),
-+
+            )
         };
         super::build_pool(connector_factory, config, partitions, policy)
-+
+    }
-+
+}
-+
 impl Builder<TlsProviderSelected> {
     /// Set the TLS context (custom trust store, etc.).
     pub fn tls_context(mut self, context: TlsContext) -> Self {
         self.tls.context = context;
         self
-+
+    }
-+
     /// This is the mutable version of [`tls_context`](Self::tls_context).
     pub fn set_tls_context(&mut self, context: TlsContext) -> &mut Self {
         self.tls.context = context;
         self
-+
+    }
-+
     /// Build an HTTPS client with the selected TLS provider.
     pub fn build_https(mut self) -> super::SharedPool {
         let dns_resolver = self.dns_resolver.take();
         let keepalive = resolve_tcp_keepalive(self.tcp_keepalive);
         match dns_resolver {
             Some(resolver) => {
                 let mut tcp = HyperHttpConnector::new_with_resolver(HyperUtilResolver { resolver });
                 tcp.set_nodelay(self.tcp_nodelay);
                 tcp.set_keepalive(keepalive);
                 tcp.enforce_http(false);
                 self.build_from_tcp(tcp)
-+
+            }
             None => {
                 let mut tcp = HyperHttpConnector::new();
                 tcp.set_nodelay(self.tcp_nodelay);
                 tcp.set_keepalive(keepalive);
                 tcp.enforce_http(false);
                 self.build_from_tcp(tcp)
-+
+            }
-+
+        }
-+
+    }
-+
     fn build_from_tcp<R>(self, tcp: HyperHttpConnector<R>) -> super::SharedPool
     where
         R: Clone + Send + Sync + 'static,
         R: tower::Service<DnsName>,
         R::Response: Iterator<Item = std::net::SocketAddr>,
         R::Future: Send,
         R::Error: Into<BoxError>,
-+
+    {
         let config = PoolConfig {
             max_connections: self.max_connections,
             max_connections_per_host: self.max_connections_per_host,
             pool_idle_timeout: resolve_pool_idle_timeout(self.pool_idle_timeout),
             connection_event_listener: self.connection_event_listener.clone(),
         };
-+
         let proxy_config = self
             .proxy_config
             .clone()
             .unwrap_or_else(ProxyConfig::disabled);
         let proxy_matcher = proxy_matcher_from(&self.proxy_config);
         let partitions = self.partitions;
         let policy = self.cross_partition_policy;
-+
         match &self.tls.provider {
             #[cfg(any(
                 feature = "rustls-aws-lc",
                 feature = "rustls-aws-lc-fips",
                 feature = "rustls-ring"
             ))]
             tls::Provider::Rustls(crypto_mode) => {
                 let crypto_mode = crypto_mode.clone();
                 let tls_context = self.tls.context.clone();
                 let connector_factory = move |partition: &Partition| {
                     tls::rustls_provider::build_connector::wrap_connector(
                         bind_interface(&tcp, partition.nic.as_deref()),
                         crypto_mode.clone(),
                         &tls_context,
                         proxy_config.clone(),
-+
+                    )
                 };
                 let pool = super::build_pool(connector_factory, config, partitions, policy);
                 super::SharedPool {
                     inner: Arc::new(super::SharedPoolInner {
                         pool: Arc::new(pool),
                         proxy_matcher,
                     }),
-+
+                }
-+
+            }
             #[cfg(feature = "s2n-tls")]
             tls::Provider::S2nTls => {
                 let tls_context = self.tls.context.clone();
                 let connector_factory = move |partition: &Partition| {
                     tls::s2n_tls_provider::build_connector::wrap_connector(
                         bind_interface(&tcp, partition.nic.as_deref()),
                         &tls_context,
                         proxy_config.clone(),
-+
+                    )
                 };
                 let pool = super::build_pool(connector_factory, config, partitions, policy);
                 super::SharedPool {
                     inner: Arc::new(super::SharedPoolInner {
                         pool: Arc::new(pool),
                         proxy_matcher,
                     }),
-+
+                }
-+
+            }
             // Provider is #[non_exhaustive]; this arm is unreachable when any
             // TLS feature is enabled (which is required to construct a Provider).
             #[allow(unreachable_patterns)]
             _ => unreachable!("a TLS feature must be enabled to use build_https()"),
-+
+        }
-+
+    }
-+
+}
-+
 /// Resolve the configured pool idle timeout. Outer `None` applies the
 /// default; `Some(None)` disables eviction; `Some(Some(d))` uses `d`.
 fn resolve_pool_idle_timeout(configured: Option<Option<Duration>>) -> Option<Duration> {
     match configured {
         None => Some(DEFAULT_POOL_IDLE_TIMEOUT),
         Some(inner) => inner,
-+
+    }
-+
+}
-+
 /// Resolve the configured TCP keepalive idle time. Keepalive is disabled
 /// by default, so the outer `None` resolves to off. `Some(None)` is also
 /// off; `Some(Some(d))` enables keepalive with idle time `d`.
 fn resolve_tcp_keepalive(configured: Option<Option<Duration>>) -> Option<Duration> {
     configured.flatten()
-+
+}
-+
 /// Bind a clone of the base TCP connector to a network interface.
 ///
 /// The single place per-partition NIC binding happens: each partition's
 /// connector factory clones the shared base connector and routes it
 /// through here with that partition's `nic`. `None` (the default,
 /// no-interface case) returns an unbound clone. `set_interface` is only
 /// available on Linux-like targets; elsewhere the `nic` is accepted and
 /// ignored, matching the v1 client.
 fn bind_interface<R>(base: &HyperHttpConnector<R>, nic: Option<&str>) -> HyperHttpConnector<R>
 where
     R: Clone,
-+
+{
     #[allow(unused_mut)]
     let mut tcp = base.clone();
     #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
     if let Some(interface) = nic {
         tcp.set_interface(interface);
-+
+    }
     let _ = nic;
     tcp
-+
+}
-+
 /// Build the proxy URL matcher from a `ProxyConfig`, returning `None` when
 /// no proxy is configured. Wrapped in `Arc` for shared ownership: the pool
 /// stores it once and each `PooledConnector` clones the handle.
 pub(super) fn proxy_matcher_from(proxy_config: &Option<ProxyConfig>) -> Option<Arc<ProxyMatcher>> {
     proxy_config
         .as_ref()
         .map(|c| Arc::new(c.clone().into_hyper_util_matcher()))
-+
+}
-+
 /// Returns the port if it is not the default for the scheme.
 pub(super) fn get_non_default_port(uri: &http_1x::Uri) -> Option<http_1x::uri::Port<&str>> {
     match (uri.port().map(|p| p.as_u16()), uri.scheme()) {
         (Some(443), Some(s)) if *s == http_1x::uri::Scheme::HTTPS => None,
         (Some(80), Some(s)) if *s == http_1x::uri::Scheme::HTTP => None,
         _ => uri.port(),
-+
+    }
-+
+}
-+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::client::pool::PartitionId;
-+
     #[test]
     fn builder_defaults() {
         let b = Builder::default();
         // Outer None = "not configured"; the default is applied at build time.
         assert_eq!(b.pool_idle_timeout, None);
         assert_eq!(b.tcp_keepalive, None);
         assert!(b.tcp_nodelay, "tcp_nodelay defaults to true");
         assert_eq!(b.max_connections, None);
         assert_eq!(b.max_connections_per_host, None);
         assert!(b.proxy_config.is_none());
         assert!(b.connection_event_listener.is_none());
         assert_eq!(b.cross_partition_policy, CrossPartitionPolicy::Never);
         assert!(b.dns_resolver.is_none());
         assert!(b.partitions.is_empty());
-+
+    }
-+
     #[test]
     fn idle_timeout_resolution() {
         // Unconfigured → default applied.
         assert_eq!(
             resolve_pool_idle_timeout(None),
             Some(DEFAULT_POOL_IDLE_TIMEOUT)
         );
         // Some(None) → explicitly disabled.
         assert_eq!(resolve_pool_idle_timeout(Some(None)), None);
         // Some(Some(d)) → overridden.
         let d = Duration::from_secs(5);
         assert_eq!(resolve_pool_idle_timeout(Some(Some(d))), Some(d));
-+
+    }
-+
     #[test]
     fn keepalive_resolution() {
         // Unset → off (keepalive is disabled by default).
         assert_eq!(resolve_tcp_keepalive(None), None);
         assert_eq!(resolve_tcp_keepalive(Some(None)), None);
         let d = Duration::from_secs(45);
         assert_eq!(resolve_tcp_keepalive(Some(Some(d))), Some(d));
-+
+    }
-+
     #[test]
     fn idle_timeout_setters_set_three_states() {
         // chaining setter: Duration → Some(Some(d))
         let b = Builder::default().pool_idle_timeout(Duration::from_secs(7));
         assert_eq!(b.pool_idle_timeout, Some(Some(Duration::from_secs(7))));
         // chaining setter: None → Some(None) (disable)
         let b = Builder::default().pool_idle_timeout(None);
         assert_eq!(b.pool_idle_timeout, Some(None));
         // mutable setter passes through verbatim
         let mut b = Builder::default();
         b.set_pool_idle_timeout(Some(None));
         assert_eq!(b.pool_idle_timeout, Some(None));
-+
+    }
-+
     #[test]
     fn keepalive_setters_set_three_states() {
         let b = Builder::default().tcp_keepalive(Duration::from_secs(15));
         assert_eq!(b.tcp_keepalive, Some(Some(Duration::from_secs(15))));
         let b = Builder::default().tcp_keepalive(None);
         assert_eq!(b.tcp_keepalive, Some(None));
         let mut b = Builder::default();
         b.set_tcp_keepalive(Some(None));
         assert_eq!(b.tcp_keepalive, Some(None));
-+
+    }
-+
     /// `tls_provider` transitions the type-state while preserving every
     /// configured field. Guards the hand-written field-by-field move in
     /// `tls_provider` against a dropped field on a future edit.
     #[test]
     fn tls_provider_preserves_all_config() {
         let b = Builder::default()
             .pool_idle_timeout(Duration::from_secs(30))
             .tcp_nodelay(false)
             .tcp_keepalive(Duration::from_secs(45))
             .max_connections(100)
             .max_connections_per_host(10)
             .cross_partition_policy(CrossPartitionPolicy::PreferLocal)
             .partitions([Partition::new(
                 PartitionId::from_index(0),
                 crate::client::pool::TokioDriverSpawner::from_handle(
                     // a handle is only needed to construct the spawner; no
                     // runtime work happens here.
                     tokio::runtime::Builder::new_current_thread()
                         .build()
                         .unwrap()
                         .handle()
                         .clone(),
                 ),
             )]);
-+
         let provider = tls::Provider::Rustls(tls::rustls_provider::CryptoMode::AwsLc);
         let b = b.tls_provider(provider);
-+
         assert_eq!(b.pool_idle_timeout, Some(Some(Duration::from_secs(30))));
         assert!(!b.tcp_nodelay);
         assert_eq!(b.tcp_keepalive, Some(Some(Duration::from_secs(45))));
         assert_eq!(b.max_connections, Some(100));
         assert_eq!(b.max_connections_per_host, Some(10));
         assert_eq!(b.cross_partition_policy, CrossPartitionPolicy::PreferLocal);
         assert_eq!(b.partitions.len(), 1);
-+
+    }
-+
     #[test]
     fn non_default_port_elided_per_scheme() {
         let cases = [
             ("https://example.com/", None), // 443 elided
             ("http://example.com/", None),  // 80 elided
             ("https://example.com:8443/", Some(8443)),
             ("http://example.com:8080/", Some(8080)),
             ("https://example.com:80/", Some(80)), // 80 is non-default for https
             ("http://example.com:443/", Some(443)), // 443 is non-default for http
         ];
         for (uri, expected) in cases {
             let uri: http_1x::Uri = uri.parse().unwrap();
             let port = get_non_default_port(&uri).map(|p| p.as_u16());
             assert_eq!(port, expected, "uri = {uri}");
-+
+        }
-+
+    }
-+
+}

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

@@ -0,1 +0,231 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-+
 //! Per-partition client handle.
-+
 use std::borrow::Cow;
 use std::sync::Arc;
 use std::time::Duration;
-+
 use aws_smithy_async::rt::sleep::SharedAsyncSleep;
 use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
 use aws_smithy_runtime_api::client::connector_metadata::ConnectorMetadata;
 use aws_smithy_runtime_api::client::http::{
     HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpConnector,
 };
 use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_runtime_api::client::runtime_components::RuntimeComponents;
 use hyper_util::client::proxy::matcher::Matcher as ProxyMatcher;
-+
 use super::partition::{PartitionId, PartitionState};
 use super::{ConnectionPool, SharedPool};
 use crate::client::downcast_error;
 use crate::client::proxy::add_proxy_auth_header;
-+
 /// Per-partition view of a [`SharedPool`].
 ///
 /// Implements [`HttpClient`] by routing requests through the shared
 /// connection pool. Multiple `Client` instances can reference the same
 /// pool, each targeting a distinct declared partition.
 ///
 /// Construct via [`Client::new`] (default partition) or
 /// [`Client::from_partition`] for a specific declared partition.
 ///
 /// Cloning is cheap: all fields are `Arc`-backed.
 #[derive(Clone)]
 pub struct Client {
     pool: SharedPool,
     partition: Arc<PartitionState>,
-+
+}
-+
 impl std::fmt::Debug for Client {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Client")
             .field("partition_id", &self.partition.id)
             .field("nic", &self.partition.nic)
             .finish_non_exhaustive()
-+
+    }
-+
+}
-+
 impl Client {
     /// Construct a `Client` targeting the pool's default partition (the
     /// first declared, or the anonymous partition when none were declared).
     pub fn new(pool: &SharedPool) -> Self {
         let partition = pool.inner.pool.registry().default_partition();
         Self {
             pool: pool.clone(),
             partition,
-+
+        }
-+
+    }
-+
     /// Construct a `Client` targeting a specific declared partition.
     /// Panics if `id` was not declared on the pool builder (programming
     /// error: the caller declared the topology).
     pub fn from_partition(pool: &SharedPool, id: PartitionId) -> Self {
         let partition = pool.inner.pool.registry().partition(id);
         Self {
             pool: pool.clone(),
             partition,
-+
+        }
-+
+    }
-+
     /// The partition id this client targets.
     #[cfg(test)]
     fn partition_id(&self) -> PartitionId {
         self.partition.id
-+
+    }
-+
+}
-+
 impl HttpClient for Client {
     fn http_connector(
         &self,
         settings: &HttpConnectorSettings,
         components: &RuntimeComponents,
     ) -> SharedHttpConnector {
         let connect_timeout = settings.connect_timeout();
         let read_timeout = settings.read_timeout();
         let sleep_impl = components.sleep_impl();
-+
         if (connect_timeout.is_some() || read_timeout.is_some()) && sleep_impl.is_none() {
             panic!(
                 "an async sleep impl is required to use connect/read timeouts with \
                  the v2 HTTP client; provide one via `RuntimeComponents::sleep_impl`"
             );
-+
+        }
-+
         SharedHttpConnector::new(PooledConnector {
             pool: self.pool.inner.pool.clone(),
             partition: self.partition.clone(),
             connect_timeout,
             read_timeout,
             sleep_impl,
             proxy_matcher: self.pool.inner.proxy_matcher.clone(),
         })
-+
+    }
-+
     fn connector_metadata(&self) -> Option<ConnectorMetadata> {
         Some(ConnectorMetadata::new("hyper", Some(Cow::Borrowed("1.x"))))
-+
+    }
-+
+}
-+
 // ---------------------------------------------------------------------------
 // PooledConnector (HttpConnector adapter)
 // ---------------------------------------------------------------------------
-+
 /// Smithy [`HttpConnector`] backed by the v2 connection pool.
 ///
 /// Constructed fresh per [`HttpClient::http_connector`] call so it can
 /// capture the per-operation [`HttpConnectorSettings`] (connect/read
 /// timeouts). The pool itself is shared across all operations via `Arc`.
 struct PooledConnector {
     pool: Arc<ConnectionPool>,
     partition: Arc<PartitionState>,
     connect_timeout: Option<Duration>,
     read_timeout: Option<Duration>,
     sleep_impl: Option<SharedAsyncSleep>,
     proxy_matcher: Option<Arc<ProxyMatcher>>,
-+
+}
-+
 impl std::fmt::Debug for PooledConnector {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("PooledConnector").finish()
-+
+    }
-+
+}
-+
 impl HttpConnector for PooledConnector {
     fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
         let pool = self.pool.clone();
         let partition = self.partition.clone();
         let connect_timeout = self.connect_timeout;
         let read_timeout = self.read_timeout;
         let sleep_impl = self.sleep_impl.clone();
         let proxy_matcher = self.proxy_matcher.clone();
         HttpConnectorFuture::new(async move {
             let mut request = request
                 .try_into_http1x()
                 .map_err(|err| ConnectorError::user(err.into()))?;
-+
             let full_uri = request.uri().clone();
-+
             if let Some(matcher) = proxy_matcher.as_ref() {
                 add_proxy_auth_header(&mut request, matcher);
-+
+            }
-+
             if let Some(capture_smithy) = request.extensions().get::<CaptureSmithyConnection>() {
                 let capture = super::ConnectionMetadataCapture::new();
                 let for_retriever = capture.clone();
                 capture_smithy.set_connection_retriever(move || for_retriever.get());
                 request.extensions_mut().insert(capture);
-+
+            }
-+
             if let Some((duration, sleep)) = read_timeout.zip(sleep_impl.clone()) {
                 request.extensions_mut().insert(super::ReadTimeoutHint(
                     super::TimeoutContext::new(duration, sleep),
                 ));
-+
+            }
-+
             if !request.headers().contains_key(http_1x::header::HOST) {
                 if let Some(authority) = full_uri.authority() {
                     let host = match super::builder::get_non_default_port(&full_uri) {
                         Some(port) => format!("{}:{}", authority.host(), port),
                         None => authority.host().to_string(),
                     };
                     request.headers_mut().insert(
                         http_1x::header::HOST,
                         http_1x::HeaderValue::from_str(&host)
                             .expect("authority is valid header value"),
                     );
-+
+                }
-+
+            }
-+
             let connect_ctx = super::ConnectCtx::new(
                 full_uri,
                 connect_timeout
                     .zip(sleep_impl)
                     .map(|(d, s)| super::TimeoutContext::new(d, s)),
             );
-+
             let response = pool
                 .send_request(&partition, connect_ctx, request)
                 .await
                 .map_err(downcast_error)?;
-+
             HttpResponse::try_from(response).map_err(|err| ConnectorError::other(err.into(), None))
         })
-+
+    }
-+
+}
-+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::client::pool::partition::{Partition, TokioDriverSpawner};
-+
     #[tokio::test]
     async fn client_new_uses_default_partition() {
         let pool = SharedPool::builder().build_http();
         let client = Client::new(&pool);
         assert_eq!(client.partition_id(), PartitionId::default());
-+
+    }
-+
     #[tokio::test]
     async fn from_partition_resolves_declared() {
         let pool = SharedPool::builder()
             .partitions([Partition::new(
                 PartitionId::from_index(3),
                 TokioDriverSpawner::current(),
             )])
             .build_http();
         let client = Client::from_partition(&pool, PartitionId::from_index(3));
         assert_eq!(client.partition_id(), PartitionId::from_index(3));
-+
+    }
-+
     #[tokio::test]
     #[should_panic(expected = "partition not declared")]
     async fn from_partition_unknown_panics() {
         let pool = SharedPool::builder().build_http();
         Client::from_partition(&pool, PartitionId::from_index(99));
-+
+    }
-+
+}

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

@@ -0,1 +0,1400 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-+
 //! Connection state tracking for pooled connections.
-+
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
-+
 use aws_smithy_async::rt::sleep::SharedAsyncSleep;
 use aws_smithy_runtime_api::box_error::BoxError;
 use aws_smithy_runtime_api::client::connection::{ConnectionId, ConnectionMetadata};
 use aws_smithy_types::body::SdkBody;
 use pin_project_lite::pin_project;
 use tokio::sync::OwnedSemaphorePermit;
 use tower::Service;
-+
 use super::cache;
 use super::handshake::H1SendRequest;
-+
 /// Authority of a pooled connection (host with optional port).
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct Authority(Arc<str>);
-+
 impl Authority {
     pub(crate) fn new(s: impl Into<Arc<str>>) -> Self {
         Self(s.into())
-+
+    }
-+
     /// Construct an authority key for a host, as `host` or `host:port`.
     ///
     /// Used to query connection counts via [`SharedPool::stats`]. The
     /// pool keys connection state by the authority component of each
     /// request's URI, compared as an exact byte string. The value passed
     /// here must match that form for a lookup to hit: notably, a port is
     /// present only when the URI carried a non-default port (an HTTPS URI
     /// to `example.com` keys as `example.com`, not `example.com:443`), and
     /// the host is matched case-sensitively. A value that does not match
     /// any keyed authority yields empty stats rather than an error.
     ///
     /// [`SharedPool::stats`]: crate::client::pool::SharedPool::stats
     pub fn from_host(host: impl Into<Arc<str>>) -> Self {
         Self(host.into())
-+
+    }
-+
     /// The authority as a string slice.
     pub fn as_str(&self) -> &str {
         &self.0
-+
+    }
-+
+}
-+
 impl AsRef<str> for Authority {
     fn as_ref(&self) -> &str {
         &self.0
-+
+    }
-+
+}
-+
 impl std::fmt::Display for Authority {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         self.0.fmt(f)
-+
+    }
-+
+}
-+
 /// Protocol negotiated for a connection.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 #[non_exhaustive]
 pub enum NegotiatedProtocol {
     /// HTTP/1.1
     Http1,
     /// HTTP/2
     Http2,
-+
+}
-+
 /// Why a connection was removed from the pool.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 #[non_exhaustive]
 pub enum CloseReason {
     /// Idle longer than the configured pool idle timeout.
     IdleTimeout,
     /// Marked as poisoned (unhealthy) by the SDK or pool.
     Poisoned,
     /// Connection found dead at checkout (e.g., server closed the
     /// connection while it was idle in the pool). The associated error
     /// on the event carries specifics.
     Unusable,
     /// The pool itself was dropped.
     PoolDropped,
     /// Dropped to free a connection permit for another partition under
     /// cap pressure (cross-partition active reclaim). Distinct from
     /// `IdleTimeout`: the connection was still within its idle window but
     /// was reclaimed because a starved partition needed the capacity.
     Reclaimed,
-+
+}
-+
 /// Timing breakdown for connection establishment.
 #[derive(Clone, Copy, Debug)]
 #[non_exhaustive]
 pub struct ConnectionTiming {
     /// TCP connect + TLS handshake combined. Measured from connector call
     /// start to connected IO stream returned.
     connect_duration: Duration,
-+
+}
-+
 impl ConnectionTiming {
     pub(crate) fn new(connect_duration: Duration) -> Self {
         Self { connect_duration }
-+
+    }
-+
     /// Total time to establish the transport (TCP + TLS).
     pub fn connect_duration(&self) -> Duration {
         self.connect_duration
-+
+    }
-+
+}
-+
 /// Emitted when a new connection is established (TCP + TLS + HTTP handshake).
 #[derive(Debug)]
 #[non_exhaustive]
 pub struct ConnectionCreatedEvent {
     conn_id: ConnectionId,
     authority: Authority,
     remote_addr: Option<SocketAddr>,
     protocol: NegotiatedProtocol,
     timing: ConnectionTiming,
-+
+}
-+
 impl ConnectionCreatedEvent {
     pub(crate) fn new(
         conn_id: ConnectionId,
         authority: Authority,
         remote_addr: Option<SocketAddr>,
         protocol: NegotiatedProtocol,
         timing: ConnectionTiming,
     ) -> Self {
         Self {
             conn_id,
             authority,
             remote_addr,
             protocol,
             timing,
-+
+        }
-+
+    }
-+
     /// The pool-assigned connection identifier.
     pub fn conn_id(&self) -> ConnectionId {
         self.conn_id
-+
+    }
-+
     /// The authority (host:port) this connection is for.
     pub fn authority(&self) -> &Authority {
         &self.authority
-+
+    }
-+
     /// Remote address of the peer, if known.
     pub fn remote_addr(&self) -> Option<SocketAddr> {
         self.remote_addr
-+
+    }
-+
     /// Negotiated protocol.
     pub fn protocol(&self) -> NegotiatedProtocol {
         self.protocol
-+
+    }
-+
     /// Timing breakdown for connection establishment.
     pub fn timing(&self) -> &ConnectionTiming {
         &self.timing
-+
+    }
-+
+}
-+
 /// Emitted when an existing idle connection is checked out from the pool.
 #[derive(Debug)]
 #[non_exhaustive]
 pub struct ConnectionReusedEvent {
     conn_id: ConnectionId,
     authority: Authority,
-+
+}
-+
 impl ConnectionReusedEvent {
     pub(crate) fn new(conn_id: ConnectionId, authority: Authority) -> Self {
         Self { conn_id, authority }
-+
+    }
-+
     /// The pool-assigned connection identifier.
     pub fn conn_id(&self) -> ConnectionId {
         self.conn_id
-+
+    }
-+
     /// The authority (host:port) this connection is for.
     pub fn authority(&self) -> &Authority {
         &self.authority
-+
+    }
-+
+}
-+
 /// Emitted when a connection is removed from the pool.
 #[non_exhaustive]
 pub struct ConnectionClosedEvent {
     conn_id: ConnectionId,
     authority: Authority,
     remote_addr: Option<SocketAddr>,
     reason: CloseReason,
     error: Option<BoxError>,
-+
+}
-+
 impl std::fmt::Debug for ConnectionClosedEvent {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let mut s = f.debug_struct("ConnectionClosedEvent");
         s.field("conn_id", &self.conn_id)
             .field("authority", &self.authority)
             .field("remote_addr", &self.remote_addr)
             .field("reason", &self.reason);
         if let Some(ref e) = self.error {
             s.field("error", &format_args!("{e}"));
-+
+        }
         s.finish()
-+
+    }
-+
+}
-+
 impl ConnectionClosedEvent {
     pub(crate) fn new(
         conn_id: ConnectionId,
         authority: Authority,
         remote_addr: Option<SocketAddr>,
         reason: CloseReason,
         error: Option<BoxError>,
     ) -> Self {
         Self {
             conn_id,
             authority,
             remote_addr,
             reason,
             error,
-+
+        }
-+
+    }
-+
     /// The pool-assigned connection identifier.
     pub fn conn_id(&self) -> ConnectionId {
         self.conn_id
-+
+    }
-+
     /// The authority (host:port) this connection was for.
     pub fn authority(&self) -> &Authority {
         &self.authority
-+
+    }
-+
     /// Remote address of the peer, if known.
     pub fn remote_addr(&self) -> Option<SocketAddr> {
         self.remote_addr
-+
+    }
-+
     /// Why the connection was closed.
     pub fn reason(&self) -> CloseReason {
         self.reason
-+
+    }
-+
     /// The error associated with this close, if any. Present for
     /// server-initiated closes; `None` for policy-driven closes
     /// (idle timeout, poisoning).
     pub fn error(&self) -> Option<&(dyn std::error::Error + Send + Sync)> {
         self.error.as_ref().map(|e| e.as_ref())
-+
+    }
-+
+}
-+
 /// Emitted when a connection attempt fails before completing the handshake.
 #[non_exhaustive]
 pub struct ConnectionFailedEvent {
     authority: Authority,
     remote_addr: Option<SocketAddr>,
     error: BoxError,
-+
+}
-+
 impl std::fmt::Debug for ConnectionFailedEvent {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("ConnectionFailedEvent")
             .field("authority", &self.authority)
             .field("remote_addr", &self.remote_addr)
             .field("error", &format_args!("{}", self.error))
             .finish()
-+
+    }
-+
+}
-+
 impl ConnectionFailedEvent {
     pub(crate) fn new(
         authority: Authority,
         remote_addr: Option<SocketAddr>,
         error: BoxError,
     ) -> Self {
         Self {
             authority,
             remote_addr,
             error,
-+
+        }
-+
+    }
-+
     /// The authority (host:port) the connection was attempting to reach.
     pub fn authority(&self) -> &Authority {
         &self.authority
-+
+    }
-+
     /// Remote address of the peer, if known. `None` when the failure occurred
     /// before a peer address was established (e.g., DNS resolution failure or
     /// connection refused before address binding).
     pub fn remote_addr(&self) -> Option<SocketAddr> {
         self.remote_addr
-+
+    }
-+
     /// The error that caused the connection attempt to fail.
     pub fn error(&self) -> &(dyn std::error::Error + Send + Sync) {
         self.error.as_ref()
-+
+    }
-+
+}
-+
 /// Callback for connection lifecycle events within the pool.
 ///
 /// Implementations receive notifications when connections are created,
 /// reused from the pool, closed, or fail to establish.
 ///
 /// Implementations must be non-blocking. Defer expensive work to a
 /// background task.
 pub trait ConnectionEventListener: Send + Sync + 'static {
     /// A new connection was established.
     fn on_created(&self, _event: &ConnectionCreatedEvent) {}
     /// An existing idle connection was checked out from the pool.
     fn on_reused(&self, _event: &ConnectionReusedEvent) {}
     /// A connection was removed from the pool.
     fn on_closed(&self, _event: &ConnectionClosedEvent) {}
     /// A connection attempt failed before completing the handshake.
     fn on_connection_failed(&self, _event: &ConnectionFailedEvent) {}
-+
+}
-+
 /// A duration paired with the sleep implementation used to realize it.
 ///
 /// This type makes "timeout without sleep impl" unrepresentable: you cannot
 /// construct one without committing to a way to actually wait. Created at the
 /// adapter layer from `HttpConnectorSettings` + `RuntimeComponents::sleep_impl()`
 /// and passed down the pool stack where timeouts are applied.
 #[derive(Clone, Debug)]
 pub(crate) struct TimeoutContext {
     pub(crate) duration: Duration,
     pub(crate) sleep_impl: SharedAsyncSleep,
-+
+}
-+
 impl TimeoutContext {
     pub(crate) fn new(duration: Duration, sleep_impl: SharedAsyncSleep) -> Self {
         Self {
             duration,
             sleep_impl,
-+
+        }
-+
+    }
-+
+}
-+
 /// Target type for the connect portion of the pool stack.
 ///
 /// Replaces bare `Uri` so per-operation connect metadata flows through the
 /// composable pool types (Map → Negotiate → Cache → handshake → ConnectionLimit
 /// → TCP connector) to the layers that need them.
 #[derive(Clone, Debug)]
 pub(crate) struct ConnectCtx {
     /// Target URI: its authority is the pool key, and the full URI is
     /// passed to the TCP connector.
     pub(crate) uri: http_1x::Uri,
     /// Bounds new-connection establishment (TCP + TLS handshake). `None`
     /// means no connect timeout; cached connections skip the connector
     /// entirely so this is automatically a no-op on cache hit.
     pub(crate) connect_timeout: Option<TimeoutContext>,
     /// How the connect path behaves when a permit cannot be acquired.
     pub(crate) mode: AcquireMode,
-+
+}
-+
 /// Behavior of the connect path when the connection cap is reached.
 ///
 /// Selected per request (carried on [`ConnectCtx`]) because the same
 /// partition stack is exercised twice under `PreferLocal`: once
 /// `NonBlocking` to probe for local capacity, then `Blocking` as the
 /// fallback after a peer-borrow miss.
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(crate) enum AcquireMode {
     /// Block on the semaphore until a permit is free (active reclaim, then
     /// FIFO wait). The authoritative take.
     #[default]
     Blocking,
     /// Return [`CapBound`] immediately on `NoPermits` instead of blocking,
     /// so the caller can try borrowing a peer's connection first.
     NonBlocking,
-+
+}
-+
 /// Sentinel error from the connect path under [`AcquireMode::NonBlocking`]:
 /// the connection cap is reached and no permit is available. Distinct from
 /// any real connect failure, and distinct from negotiate's internal
 /// `UseOther` sentinel, so it propagates verbatim up the stack.
 #[derive(Debug)]
 pub(crate) struct CapBound;
-+
 impl std::fmt::Display for CapBound {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("connection cap reached (non-blocking acquire)")
-+
+    }
-+
+}
-+
 impl std::error::Error for CapBound {}
-+
 impl CapBound {
     /// Whether `err` is a `CapBound` sentinel anywhere in its chain.
     pub(crate) fn is(err: &(dyn std::error::Error + 'static)) -> bool {
         let mut e: Option<&(dyn std::error::Error + 'static)> = Some(err);
         while let Some(cur) = e {
             if cur.is::<CapBound>() {
                 return true;
-+
+            }
             e = cur.source();
-+
+        }
         false
-+
+    }
-+
+}
-+
 impl ConnectCtx {
     pub(crate) fn new(uri: http_1x::Uri, connect_timeout: Option<TimeoutContext>) -> Self {
         Self {
             uri,
             connect_timeout,
             mode: AcquireMode::Blocking,
-+
+        }
-+
+    }
-+
     /// Set the acquire mode (defaults to [`AcquireMode::Blocking`]).
     pub(crate) fn with_mode(mut self, mode: AcquireMode) -> Self {
         self.mode = mode;
         self
-+
+    }
-+
+}
-+
 /// Request extension set by the adapter to hint a read timeout to the
 /// checkout services (`H{1,2}Checkout`).
 ///
 /// `Some(...)` means: once the connection is selected (cache hit or fresh
 /// handshake), wrap `conn.call(req)` with this timeout. Bounds request-write
 /// + response-headers-wait only. Does NOT include pool acquire or connect
 /// establishment (those have their own timeouts).
 #[derive(Clone, Debug)]
 pub(crate) struct ReadTimeoutHint(pub(crate) TimeoutContext);
-+
 /// Permits acquired from connection limit semaphores.
 /// Held for the lifetime of the connection; dropped when the connection is dropped.
 pub(crate) struct ConnectionPermit {
     _global: Option<OwnedSemaphorePermit>,
     _per_host: Option<OwnedSemaphorePermit>,
-+
+}
-+
 impl ConnectionPermit {
     pub(crate) fn new(
         global: Option<OwnedSemaphorePermit>,
         per_host: Option<OwnedSemaphorePermit>,
     ) -> Self {
         Self {
             _global: global,
             _per_host: per_host,
-+
+        }
-+
+    }
-+
+}
-+
 /// Outcome of establishing a new transport connection (TCP + TLS).
 ///
 /// Carries the IO handle and the connection permit acquired during
 /// connection establishment. The IO handle drives the protocol
 /// handshake; the permit holds the connection's slot in the pool's
 /// limit semaphores until the connection is dropped.
 pub(crate) struct EstablishedConnection<IO> {
     pub(crate) io: IO,
     pub(crate) permit: Arc<ConnectionPermit>,
     pub(crate) establishing: super::stats::EstablishingGuard,
-+
+}
-+
 /// Error returned by `ManagedConnection::poll_ready` when the connection
 /// has been marked poisoned and should not be reused.
 #[derive(Debug)]
 pub(crate) struct PoisonedError;
-+
 impl std::fmt::Display for PoisonedError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("connection poisoned")
-+
+    }
-+
+}
-+
 impl std::error::Error for PoisonedError {}
-+
 /// Metadata about a connection captured at establishment time.
 ///
 /// Captured between TLS connector output and protocol handshake (the last point
 /// where the raw transport stream is accessible).
 #[derive(Debug, Clone)]
 pub(crate) struct ConnectionInfo {
     /// Remote address of the peer. `None` when the underlying connector
     /// did not attach `HttpInfo` to its `Connected` extras.
     pub(crate) remote_addr: Option<SocketAddr>,
     /// Local address of this end of the connection.
     pub(crate) local_addr: Option<SocketAddr>,
     /// `true` when this connection is to an HTTP proxy server (rather than
     /// directly to the origin). Drives request-target form selection: H1
     /// requests dispatched on a proxied connection use absolute-form URIs;
     /// direct connections use origin-form. Populated from
     /// [`hyper_util::client::legacy::connect::Connected::is_proxied`] at
     /// handshake.
     pub(crate) is_proxied: bool,
     /// The authority (host:port) this connection is for. Populated from the
     /// URI at handshake time.
     pub(crate) authority: Authority,
-+
+}
-+
 /// A one-shot "this connection is dead, don't reuse it" flag.
 ///
 /// Shared via `Arc`; all clones observe and control the same flag.
 /// `ManagedConnection` holds one and hands out clones (via `metadata()`)
 /// that let the adapter layer mark the connection poisoned through
 /// smithy's `ConnectionMetadata::poison_fn`. On next checkout or return,
 /// the pool sees the flag set and drops the connection instead of
 /// reusing it.
 #[derive(Debug, Clone, Default)]
 pub(crate) struct PoisonPill {
     flag: Arc<AtomicBool>,
-+
+}
-+
 impl PoisonPill {
     /// Create a fresh, non-poisoned pill.
     pub(crate) fn healthy() -> Self {
         Self::default()
-+
+    }
-+
     /// Mark the connection as poisoned.
     pub(crate) fn poison(&self) {
         self.flag.store(true, Ordering::Release);
-+
+    }
-+
     /// Whether the connection has been poisoned.
     pub(crate) fn is_poisoned(&self) -> bool {
         self.flag.load(Ordering::Acquire)
-+
+    }
-+
+}
-+
 /// A connection with SDK-owned lifecycle metadata.
 ///
 /// Wraps the inner service (typically `SendRequest<SdkBody>`) with state needed
 /// for pool management: poisoning, connection identity, and permit lifetime.
 ///
 /// Clone is supported when the inner service is Clone (e.g., HTTP/2 multiplexed
 /// connections). Clones share the same poison pill and connection info, so
 /// poisoning one clone poisons all of them.
 pub(crate) struct ManagedConnection<S> {
     inner: S,
     pub(crate) info: ConnectionInfo,
     /// Stable identifier for this physical connection, unique within the
     /// owning pool. Shared across `Clone`s (an H2 connection's multiplexed
     /// request handles all carry the same `conn_id`). Used in tracing and
     /// surfaced through `ConnectionMetadata` for cross-layer correlation.
     conn_id: ConnectionId,
     created_at: Instant,
     /// Timestamp the connection last became idle (or its creation time, if
     /// it has never been returned to the pool).
     ///
     /// - **H1**: stamped by [`CachedConnection::drop`] on the unpoisoned
     ///   return-to-pool path. Presence in the cache implies idle.
     /// - **H2**: stamped by [`SingletonConnection::drop`] on the
     ///   `active_streams` 1 → 0 transition. Combined with
     ///   `active_streams > 0` in the H2 retain predicate, this prevents
     ///   eviction of a multiplexed connection that's still serving streams.
     ///
     /// `Arc<Mutex<_>>` because all `Clone`s of a `ManagedConnection`
     /// observe the same timestamp (last-write-wins). `Mutex` over an
     /// atomic-encoded `u64` because writes happen at most once per
     /// request and reads at most once per eviction tick.
     idle_at: Arc<Mutex<Instant>>,
     /// In-flight request count, used by the H2 retain predicate to keep
     /// actively-multiplexed connections alive regardless of `idle_at`.
     ///
     /// - **H1**: never incremented (always 0). H1 is serial; the cache's
     ///   idle set is the authoritative idleness signal.
     /// - **H2**: incremented by [`SingletonConnection::call`] on dispatch,
     ///   decremented by [`SingletonConnection::drop`] when the body guard
     ///   releases.
     ///
     /// Arc-shared so all `Clone`s of an H2 `ManagedConnection` mutate the
     /// same counter.
     active_streams: Arc<std::sync::atomic::AtomicUsize>,
     poison: PoisonPill,
     _permit: Arc<ConnectionPermit>,
     /// Fires `established--` when the last clone drops. Arc-shared so H2
     /// clones share one guard that drops once.
     _established: Arc<super::stats::EstablishedGuard>,
-+
+}
-+
 impl<S> ManagedConnection<S> {
     /// Create a new managed connection wrapping the given service.
     pub(crate) fn new(
         inner: S,
         info: ConnectionInfo,
         conn_id: ConnectionId,
         permit: Arc<ConnectionPermit>,
         established: super::stats::EstablishedGuard,
     ) -> Self {
         let now = Instant::now();
         Self {
             inner,
             info,
             conn_id,
             created_at: now,
             idle_at: Arc::new(Mutex::new(now)),
             active_streams: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
             poison: PoisonPill::healthy(),
             _permit: permit,
             _established: Arc::new(established),
-+
+        }
-+
+    }
-+
     /// Stable identifier for this physical connection. Shared across
     /// clones (H2 multiplexing).
     pub(crate) fn conn_id(&self) -> ConnectionId {
         self.conn_id
-+
+    }
-+
     /// Whether this connection has been poisoned.
     pub(crate) fn is_poisoned(&self) -> bool {
         self.poison.is_poisoned()
-+
+    }
-+
     /// Connection info (remote/local addresses).
     #[allow(dead_code)] // accessor for telemetry/debugging consumers
     pub(crate) fn info(&self) -> &ConnectionInfo {
         &self.info
-+
+    }
-+
     /// When this connection was established.
     #[allow(dead_code)] // accessor for connection age (telemetry/debugging)
     pub(crate) fn created_at(&self) -> Instant {
         self.created_at
-+
+    }
-+
     /// Timestamp of the last return-to-idle (or creation).
     ///
     /// Guaranteed meaningful for any connection sitting in the pool: the
     /// initial value is `created_at`, and every return-to-pool overwrites
     /// it via [`Self::mark_idle`].
     pub(crate) fn idle_at(&self) -> Instant {
         *self.idle_at.lock().expect("idle_at lock poisoned")
-+
+    }
-+
     /// Stamp the return-to-idle moment. Called by [`CachedConnection::drop`]
     /// on the unpoisoned path.
     pub(crate) fn mark_idle(&self) {
         *self.idle_at.lock().expect("idle_at lock poisoned") = Instant::now();
-+
+    }
-+
     /// Current in-flight stream count. See [`Self::active_streams`]
     /// (field doc) for semantics.
     pub(crate) fn active_streams_count(&self) -> usize {
         self.active_streams
             .load(std::sync::atomic::Ordering::Acquire)
-+
+    }
-+
     /// Clone of the Arc'd stream counter, for publication through the H2
     /// side-channel so `SingletonConnection` can increment/decrement it
     /// without reaching through the opaque `Singled<…>`.
     pub(crate) fn active_streams_ref(&self) -> Arc<std::sync::atomic::AtomicUsize> {
         self.active_streams.clone()
-+
+    }
-+
     /// Clone of the Arc'd idle timestamp, for publication through the H2
     /// side-channel so `SingletonConnection::drop` can stamp it on the
     /// `active_streams` 1 → 0 transition.
     pub(crate) fn idle_at_ref(&self) -> Arc<Mutex<Instant>> {
         self.idle_at.clone()
-+
+    }
-+
     /// Mutable access to the inner service.
     pub(crate) fn inner_mut(&mut self) -> &mut S {
         &mut self.inner
-+
+    }
-+
     /// Build a smithy `ConnectionMetadata` for this connection.
     ///
     /// The returned metadata captures a clone of the `PoisonPill`, so
     /// calling `ConnectionMetadata::poison()` flips this connection's
     /// poison flag (the same flag the pool checks on checkout/return).
     /// Address fields are copied.
     pub(crate) fn metadata(&self) -> ConnectionMetadata {
         let poison = self.poison.clone();
         let conn_id = self.conn_id;
         let remote = self.info.remote_addr;
         let mut builder = ConnectionMetadata::builder()
             .proxied(self.info.is_proxied)
             .connection_id(self.conn_id)
             .poison_fn(move || {
                 tracing::debug!(conn_id = %conn_id, ?remote, "pool: connection poisoned");
                 poison.poison();
             });
         builder
             .set_remote_addr(self.info.remote_addr)
             .set_local_addr(self.info.local_addr);
         builder.build()
-+
+    }
-+
     /// `true` when the underlying connection is to an HTTP proxy. Used by
     /// H1 dispatch to choose absolute-form URIs over origin-form.
     pub(crate) fn is_proxied(&self) -> bool {
         self.info.is_proxied
-+
+    }
-+
+}
-+
 impl<S: Clone> Clone for ManagedConnection<S> {
     fn clone(&self) -> Self {
         Self {
             inner: self.inner.clone(),
             info: self.info.clone(),
             conn_id: self.conn_id,
             created_at: self.created_at,
             idle_at: self.idle_at.clone(),
             active_streams: self.active_streams.clone(),
             poison: self.poison.clone(),
             _permit: self._permit.clone(),
             _established: self._established.clone(),
-+
+        }
-+
+    }
-+
+}
-+
 impl<S> Service<http_1x::Request<SdkBody>> for ManagedConnection<S>
 where
     S: Service<http_1x::Request<SdkBody>>,
     S::Error: Into<BoxError>,
     S::Future: Send + 'static,
     S::Response: 'static,
-+
+{
     type Response = S::Response;
     type Error = BoxError;
     type Future = std::pin::Pin<
         Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
     >;
-+
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         if self.is_poisoned() {
             return Poll::Ready(Err(PoisonedError.into()));
-+
+        }
         self.inner_mut().poll_ready(cx).map_err(Into::into)
-+
+    }
-+
     fn call(&mut self, req: http_1x::Request<SdkBody>) -> Self::Future {
         let fut = self.inner_mut().call(req);
         Box::pin(async move { fut.await.map_err(Into::into) })
-+
+    }
-+
+}
-+
 /// A connection checked out from the H1 cache.
 ///
 /// Wraps `cache::Cached<ManagedConnection<S>>` so that poisoned connections
 /// are dropped from the pool (via `Cached::discard`) instead of being
 /// returned for reuse when dropped. Healthy connections return to the pool
 /// normally through `Cached::Drop`.
 pub(crate) struct CachedConnection<S> {
     inner: Option<cache::Cached<ManagedConnection<S>>>,
     listener: Option<Arc<dyn ConnectionEventListener>>,
     counters: Arc<super::stats::ConnectionCounters>,
-+
+}
-+
 impl<S> CachedConnection<S> {
     pub(crate) fn new(
         cached: cache::Cached<ManagedConnection<S>>,
         listener: Option<Arc<dyn ConnectionEventListener>>,
         counters: Arc<super::stats::ConnectionCounters>,
     ) -> Self {
         counters.incr_active();
         let is_reuse = *cached.inner().idle_at.lock().unwrap() > cached.inner().created_at;
         if is_reuse {
             let conn_id = cached.inner().conn_id();
             let authority = cached.inner().info.authority.clone();
             tracing::trace!(conn_id = %conn_id, "pool: connection reused");
             if let Some(ref l) = listener {
                 l.on_reused(&ConnectionReusedEvent::new(conn_id, authority));
-+
+            }
-+
+        }
         Self {
             inner: Some(cached),
             listener,
             counters,
-+
+        }
-+
+    }
-+
     /// Build a smithy `ConnectionMetadata` for the underlying H1 connection.
     ///
     /// See [`ManagedConnection::metadata`].
     ///
     /// Panics if called after the inner cached handle has been consumed.
     pub(crate) fn metadata(&self) -> ConnectionMetadata {
         self.inner
             .as_ref()
             .expect("CachedConnection metadata after drop")
             .inner()
             .metadata()
-+
+    }
-+
     /// Forwards [`ManagedConnection::is_proxied`].
     pub(crate) fn is_proxied(&self) -> bool {
         self.inner
             .as_ref()
             .expect("CachedConnection is_proxied after drop")
             .inner()
             .is_proxied()
-+
+    }
-+
     /// Forwards [`ManagedConnection::conn_id`].
     pub(crate) fn conn_id(&self) -> ConnectionId {
         self.inner
             .as_ref()
             .expect("CachedConnection conn_id after drop")
             .inner()
             .conn_id()
-+
+    }
-+
+}
-+
 impl<S, Req> Service<Req> for CachedConnection<S>
 where
     cache::Cached<ManagedConnection<S>>: Service<Req>,
-+
+{
     type Response = <cache::Cached<ManagedConnection<S>> as Service<Req>>::Response;
     type Error = <cache::Cached<ManagedConnection<S>> as Service<Req>>::Error;
     type Future = <cache::Cached<ManagedConnection<S>> as Service<Req>>::Future;
-+
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         match self.inner.as_mut().unwrap().poll_ready(cx) {
             Poll::Ready(Err(e)) => {
                 // Connection is dead. If poisoned, Drop handles the event.
                 // If not poisoned, this is an unusable connection (server
                 // closed it while idle, driver died, etc.).
                 let managed = self.inner.as_ref().unwrap().inner();
                 if !managed.is_poisoned() {
                     if let Some(ref l) = self.listener {
                         l.on_closed(&ConnectionClosedEvent::new(
                             managed.conn_id,
                             managed.info.authority.clone(),
                             managed.info.remote_addr,
                             CloseReason::Unusable,
                             None,
                         ));
-+
+                    }
-+
+                }
                 Poll::Ready(Err(e))
-+
+            }
             other => other,
-+
+        }
-+
+    }
-+
     fn call(&mut self, req: Req) -> Self::Future {
         self.inner.as_mut().unwrap().call(req)
-+
+    }
-+
+}
-+
 impl<S> Drop for CachedConnection<S> {
     fn drop(&mut self) {
         self.counters.decr_active();
         if let Some(cached) = self.inner.take() {
             let managed = cached.inner();
             let conn_id = managed.conn_id;
             if managed.is_poisoned() {
                 tracing::debug!(conn_id = %conn_id, "pool: connection discarded (poisoned)");
                 if let Some(ref listener) = self.listener {
                     listener.on_closed(&ConnectionClosedEvent::new(
                         managed.conn_id(),
                         managed.info.authority.clone(),
                         managed.info.remote_addr,
                         CloseReason::Poisoned,
                         None,
                     ));
-+
+                }
                 cached.discard();
             } else {
                 managed.mark_idle();
                 tracing::trace!(conn_id = %conn_id, "pool: connection returned to idle");
-+
+            }
-+
+        }
-+
+    }
-+
+}
-+
 /// State an H2 checkout (`SingletonConnection`) needs but cannot reach
 /// through the opaque `Singled<…>` to read from the underlying
 /// `ManagedConnection`. Published by the H2 handshake on each new
 /// connection; consumed by `SingletonConnection::new` as a snapshot.
 ///
 /// `active_streams` and `idle_at` are clones of the Arcs held by the
 /// `ManagedConnection`, so updates through this state are visible to the
 /// retain predicate operating on the connection directly.
 #[derive(Clone)]
 pub(crate) struct H2ConnectionState {
     pub(crate) metadata: ConnectionMetadata,
     pub(crate) active_streams: Arc<std::sync::atomic::AtomicUsize>,
     pub(crate) idle_at: Arc<Mutex<Instant>>,
-+
+}
-+
 /// Per-host side-channel between H2 handshake (writer) and H2 checkout
 /// (reader).
 ///
 /// `hyper_util::client::pool::singleton::Singled<…>` is opaque, so the
 /// checkout side cannot reach through it to the underlying
 /// `ManagedConnection`. This ref carries everything the checkout side
 /// needs (stamped fresh on each handshake): the user-facing
 /// `ConnectionMetadata` for poison and address surfacing, plus the
 /// Arc-shared `active_streams` counter and `idle_at` timestamp that the
 /// retain predicate consults and `SingletonConnection` mutates.
 ///
 /// On re-handshake (poisoning, GOAWAY, etc.), the entire state is
 /// replaced last-writer-wins. A `SingletonConnection` constructed against
 /// the previous handshake holds its own snapshot, so its
 /// increments/decrements still target the right counter.
 #[derive(Clone, Default)]
 pub(crate) struct H2ConnectionRef {
     inner: Arc<std::sync::Mutex<Option<H2ConnectionState>>>,
-+
+}
-+
 impl H2ConnectionRef {
     pub(crate) fn new() -> Self {
         Self::default()
-+
+    }
-+
     /// Replace the current state with a freshly-handshaked connection's
     /// state (last-writer-wins).
     pub(crate) fn publish(&self, state: H2ConnectionState) {
         *self.inner.lock().unwrap() = Some(state);
-+
+    }
-+
     /// Snapshot of the published state. `None` until the first
     /// successful handshake for this host.
     pub(crate) fn current(&self) -> Option<H2ConnectionState> {
         self.inner.lock().unwrap().clone()
-+
+    }
-+
+}
-+
 /// RAII guard for one in-flight dispatch against an H2 connection.
 ///
 /// Existence reflects "this checkout has dispatched a request; the
 /// connection is busy on its behalf." Constructed by
 /// [`SingletonConnection::call`] when a request is dispatched; dropped
 /// when the response body's guard releases.
 ///
 /// Drop decrements `active_streams` and, on the 1 → 0 transition, stamps
 /// `idle_at` (the moment the connection becomes truly idle).
 struct DispatchGuard {
     active_streams: Arc<std::sync::atomic::AtomicUsize>,
     idle_at: Arc<Mutex<Instant>>,
     counters: Arc<super::stats::ConnectionCounters>,
-+
+}
-+
 impl DispatchGuard {
     /// Start a dispatch against the connection described by `state`.
     /// Increments `active_streams`; the returned guard releases the
     /// increment on drop.
     fn start(state: &H2ConnectionState, counters: Arc<super::stats::ConnectionCounters>) -> Self {
         state
             .active_streams
             .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
         counters.incr_active();
         Self {
             active_streams: state.active_streams.clone(),
             idle_at: state.idle_at.clone(),
             counters,
-+
+        }
-+
+    }
-+
+}
-+
 impl Drop for DispatchGuard {
     fn drop(&mut self) {
         self.counters.decr_active();
         let prev = self
             .active_streams
             .fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
         if prev == 1 {
             if let Ok(mut idle_at) = self.idle_at.lock() {
                 *idle_at = Instant::now();
-+
+            }
-+
+        }
-+
+    }
-+
+}
-+
 /// A checked-out H2 connection ready to dispatch requests.
 ///
 /// H2 is multiplexed: the underlying `ManagedConnection` stays resident in
 /// `Singleton` for the duration of every concurrent request, so "presence
 /// in a cache" (the idleness signal we use for H1) doesn't apply.
 /// Instead, `SingletonConnection` mints a [`DispatchGuard`] on each
 /// `call`; the guard's lifetime tracks one in-flight request and its
 /// drop releases the corresponding `active_streams` slot, stamping
 /// `idle_at` on the final release. The H2 retain predicate keeps the
 /// connection alive while `active_streams > 0` regardless of `idle_at`.
 ///
 /// Lifecycle:
 /// - Constructed in the H2 upgrade `map_response` after `Singleton::call`
 ///   completes. Snapshots the current [`H2ConnectionState`] and holds it
 ///   for the checkout's lifetime.
 /// - [`Self::call`] starts a [`DispatchGuard`] before delegating to
 ///   `Singled::call`. The guard is held in `dispatch` for the rest of
 ///   this `SingletonConnection`'s lifetime.
 /// - On drop, the guard's drop releases the active-stream count and may
 ///   stamp `idle_at`. If `call` was never invoked (e.g. the surrounding
 ///   checkout was abandoned by the post-checkout `poll_ready` retry
 ///   loop), `dispatch` is `None` and drop is a no-op.
 pub(crate) struct SingletonConnection<T> {
     inner: T,
     /// Snapshot of the H2 side-channel state taken at construction.
     ///
     /// Held for the lifetime of this checkout so dispatch guards always
     /// target the connection this checkout was issued against, even if a
     /// re-handshake replaces the underlying `H2ConnectionRef`'s published
     /// state in the meantime.
     state: Option<H2ConnectionState>,
     /// Active-stream guard. `Some` between [`Self::call`] and drop.
     /// `None` if `call` was never invoked or before `call` runs.
     dispatch: Option<DispatchGuard>,
     /// Pool-level warmth counters for this (partition, authority) cell.
     counters: Arc<super::stats::ConnectionCounters>,
-+
+}
-+
 impl<T> SingletonConnection<T> {
     pub(crate) fn new(
         inner: T,
         h2_ref: H2ConnectionRef,
         counters: Arc<super::stats::ConnectionCounters>,
     ) -> Self {
         let state = h2_ref.current();
         Self {
             inner,
             state,
             dispatch: None,
             counters,
-+
+        }
-+
+    }
-+
     /// Metadata for the H2 connection this checkout was issued against.
     ///
     /// Returns `None` only if `SingletonConnection` was constructed before
     /// any handshake had published (which doesn't happen in normal flow,
     /// since `Singleton::call` always completes a handshake before
     /// yielding the `Singled` we wrap).
     pub(crate) fn metadata(&self) -> Option<ConnectionMetadata> {
         self.state.as_ref().map(|s| s.metadata.clone())
-+
+    }
-+
+}
-+
 impl<T, Req> Service<Req> for SingletonConnection<T>
 where
     T: Service<Req>,
-+
+{
     type Response = T::Response;
     type Error = T::Error;
     type Future = T::Future;
-+
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.inner.poll_ready(cx)
-+
+    }
-+
     fn call(&mut self, req: Req) -> Self::Future {
         if let Some(state) = self.state.as_ref() {
             self.dispatch = Some(DispatchGuard::start(state, self.counters.clone()));
-+
+        }
         self.inner.call(req)
-+
+    }
-+
+}
-+
 pin_project! {
     /// A response body that keeps the originating pool checkout alive until
     /// the body is fully consumed.
     ///
     /// # Why this exists
     ///
     /// When a checked-out pool connection's `Service::call` returns, the
     /// response head is available but the body may still be streaming over
     /// the same underlying HTTP connection. For H1 this is critical: if the
     /// `CachedConnection` drops at that point, the connection returns to the
     /// pool mid-body-stream and the next checkout would be handed a still-busy
     /// connection.
     ///
     /// `GuardedBody` holds the checkout (`CachedConnection` for H1,
     /// `SingletonConnection` for H2) in its guard field until the body is
     /// fully dropped. Body streaming continues through the held inner
     /// `Incoming`; when the `GuardedBody` is dropped the guard drops, which
     /// for H1 triggers `CachedConnection::Drop` (return-to-pool or `discard`
     /// if poisoned), and for H2 releases the stream's `DispatchGuard`
     /// (decrementing `active_streams`, and stamping `idle_at` on the final
     /// release).
     ///
     /// The H2 variant carries a generic type parameter because
     /// `SingletonConnection<T>`'s inner `T` is `hyper_util::client::pool::
     /// singleton::Singled<...>`, which is unnameable outside hyper-util.
     /// The H1 variant is fully concrete because `cache::Cached<...>` is
     /// nameable.
     pub(crate) struct GuardedBody<H2Unnameable> {
         #[pin]
         inner: hyper::body::Incoming,
         _guard: ConnectionGuard<H2Unnameable>,
-+
+    }
-+
+}
-+
 /// What a `GuardedBody` holds alive while the body streams.
 ///
 /// Explicit per-leg variants so H1 vs H2 bifurcation is visible at the
 /// type level and in debugger output.
 pub(crate) enum ConnectionGuard<H2Unnameable> {
     H1(CachedConnection<H1SendRequest>),
     H2(SingletonConnection<H2Unnameable>),
-+
+}
-+
 impl<H2Unnameable> GuardedBody<H2Unnameable> {
     pub(crate) fn new(inner: hyper::body::Incoming, guard: ConnectionGuard<H2Unnameable>) -> Self {
         Self {
             inner,
             _guard: guard,
-+
+        }
-+
+    }
-+
+}
-+
 impl<H2Unnameable> hyper::body::Body for GuardedBody<H2Unnameable> {
     type Data = <hyper::body::Incoming as hyper::body::Body>::Data;
     type Error = <hyper::body::Incoming as hyper::body::Body>::Error;
-+
     fn poll_frame(
         self: std::pin::Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
         self.project().inner.poll_frame(cx)
-+
+    }
-+
     fn is_end_stream(&self) -> bool {
         self.inner.is_end_stream()
-+
+    }
-+
     fn size_hint(&self) -> hyper::body::SizeHint {
         self.inner.size_hint()
-+
+    }
-+
+}
-+
 /// The response type both H1 and H2 pool checkouts produce.
 ///
 /// Carries `GuardedBody<PoolUnnameable>` so the H2 leg can hold its
 /// checkout guard (`Singled<…>`, type-erased through `PoolUnnameable`)
 /// for the response body's lifetime. `Negotiate` requires uniform
 /// response types across its legs; this alias is the uniform type that
 /// consumers above the Negotiate composition point work with.
 pub(crate) type CheckoutResponse<PoolUnnameable> = http_1x::Response<GuardedBody<PoolUnnameable>>;
-+
 /// Wire-level IO wrapper sitting below TLS in the connector stack. Pure
 /// passthrough; never modifies data or buffers.
 ///
 /// TODO(pool): instrument TCP-connect vs. TLS-handshake timing separately
 /// here (this wrapper is the seam below TLS where the TCP-only duration is
 /// observable); surface the split on `ConnectionTiming`.
 pub(crate) struct TransportIo<IO> {
     inner: IO,
-+
+}
-+
 impl<IO> TransportIo<IO> {
     fn new(inner: IO) -> Self {
         Self { inner }
-+
+    }
-+
+}
-+
 impl<IO: Unpin> Unpin for TransportIo<IO> {}
-+
 impl<IO: hyper::rt::Read + Unpin> hyper::rt::Read for TransportIo<IO> {
     fn poll_read(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
         buf: hyper::rt::ReadBufCursor<'_>,
     ) -> Poll<std::io::Result<()>> {
         std::pin::Pin::new(&mut self.inner).poll_read(cx, buf)
-+
+    }
-+
+}
-+
 impl<IO: hyper::rt::Write + Unpin> hyper::rt::Write for TransportIo<IO> {
     fn poll_write(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
         buf: &[u8],
     ) -> Poll<std::io::Result<usize>> {
         std::pin::Pin::new(&mut self.inner).poll_write(cx, buf)
-+
+    }
-+
     fn poll_flush(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> Poll<std::io::Result<()>> {
         std::pin::Pin::new(&mut self.inner).poll_flush(cx)
-+
+    }
-+
     fn poll_shutdown(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> Poll<std::io::Result<()>> {
         std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
-+
+    }
-+
+}
-+
 impl<IO: hyper_util::client::legacy::connect::Connection>
     hyper_util::client::legacy::connect::Connection for TransportIo<IO>
-+
+{
     fn connected(&self) -> hyper_util::client::legacy::connect::Connected {
         self.inner.connected()
-+
+    }
-+
+}
-+
 /// Wraps a TCP connector with the [`TransportIo`] seam below the TLS layer.
 ///
 /// The wrapper is the point at which transport-level (TCP) timing and byte
 /// accounting can be observed independently of the TLS handshake above it.
 pub(crate) struct TimingConnector<C> {
     inner: C,
-+
+}
-+
 impl<C> TimingConnector<C> {
     pub(crate) fn new(inner: C) -> Self {
         Self { inner }
-+
+    }
-+
+}
-+
 impl<C: Clone> Clone for TimingConnector<C> {
     fn clone(&self) -> Self {
         Self {
             inner: self.inner.clone(),
-+
+        }
-+
+    }
-+
+}
-+
 impl<C, IO> tower::Service<http_1x::Uri> for TimingConnector<C>
 where
     C: tower::Service<http_1x::Uri, Response = IO>,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: Send + 'static,
-+
+{
     type Response = TransportIo<IO>;
     type Error = BoxError;
     type Future = std::pin::Pin<
         Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
     >;
-+
     fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.inner.poll_ready(cx).map_err(Into::into)
-+
+    }
-+
     fn call(&mut self, uri: http_1x::Uri) -> Self::Future {
         let fut = self.inner.call(uri);
         Box::pin(async move {
             let io = fut.await.map_err(Into::into)?;
             Ok(TransportIo::new(io))
         })
-+
+    }
-+
+}
-+
 #[cfg(test)]
 mod tests {
     //! Unit tests for H2 active-stream tracking.
     //!
     //! The test harness (`ConnectionTestHarness`) is plain HTTP only, without
     //! ALPN, so the full H2 pipeline is not exercised end-to-end here. These
     //! tests verify the atomic transitions directly, which is the
     //! architectural correctness property we need: an in-flight H2 connection
     //! must not be evicted by the retain predicate, and the connection's
     //! `idle_at` must be stamped only on the `active_streams` 1 → 0
     //! transition.
     use super::*;
     use std::future::Future;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use tower::Service as _;
-+
     /// A minimal `Service<()>` that returns `Ok(())` (stand-in for
     /// `Singled::call`). We only care about the wrapper's stream-count
     /// side effects.
     #[derive(Clone, Default)]
     struct OkService;
     impl Service<()> for OkService {
         type Response = ();
         type Error = BoxError;
         type Future = std::pin::Pin<Box<dyn Future<Output = Result<(), BoxError>> + Send>>;
         fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
             Poll::Ready(Ok(()))
-+
+        }
         fn call(&mut self, _: ()) -> Self::Future {
             Box::pin(async { Ok(()) })
-+
+        }
-+
+    }
-+
     fn publish_state(h2_ref: &H2ConnectionRef) -> (Arc<AtomicUsize>, Arc<Mutex<Instant>>) {
         use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
         let active = Arc::new(AtomicUsize::new(0));
         let idle_at = Arc::new(Mutex::new(Instant::now()));
         h2_ref.publish(H2ConnectionState {
             metadata: ConnectionMetadata::builder()
                 .proxied(false)
                 .poison_fn(|| {})
                 .build(),
             active_streams: active.clone(),
             idle_at: idle_at.clone(),
         });
         (active, idle_at)
-+
+    }
-+
     /// A round-trip through `SingletonConnection::call` mints a
     /// `DispatchGuard` that releases on drop. Counter goes 0 → 1 → 0.
     #[tokio::test]
     async fn singleton_dispatch_round_trip_is_net_zero() {
         let h2_ref = H2ConnectionRef::new();
         let (active, _) = publish_state(&h2_ref);
         assert_eq!(active.load(Ordering::Acquire), 0);
-+
         let counters = Arc::new(super::super::stats::ConnectionCounters::default());
         let mut sc = SingletonConnection::new(OkService, h2_ref, counters);
         let _ = sc.call(()).await;
         assert_eq!(
             active.load(Ordering::Acquire),
 ,
             "call should mint a DispatchGuard, incrementing active_streams"
         );
         drop(sc);
         assert_eq!(
             active.load(Ordering::Acquire),
 ,
             "DispatchGuard's Drop should release the active_streams slot"
         );
-+
+    }
-+
     /// Constructing without dispatching leaves no `DispatchGuard`, so
     /// dropping is a no-op. The post-checkout `poll_ready` retry loop
     /// relies on this: a checkout discarded before `call` must not
     /// underflow the counter.
     #[tokio::test]
     async fn singleton_drop_without_dispatch_is_noop() {
         let h2_ref = H2ConnectionRef::new();
         let (active, _) = publish_state(&h2_ref);
         active.store(5, Ordering::Release);
-+
         let counters = Arc::new(super::super::stats::ConnectionCounters::default());
         let sc = SingletonConnection::new(OkService, h2_ref, counters);
         drop(sc);
         assert_eq!(
             active.load(Ordering::Acquire),
 ,
             "uncalled SingletonConnection must not release a DispatchGuard"
         );
-+
+    }
-+
     /// Concurrent dispatches against the same H2 connection (simulating
     /// multiplexed requests) all land on the same counter. `idle_at` is
     /// stamped only on the LAST `DispatchGuard` drop (the 1 → 0
     /// transition).
     #[tokio::test]
     async fn idle_at_stamped_only_on_last_dispatch_release() {
         let h2_ref = H2ConnectionRef::new();
         let (active, idle_at) = publish_state(&h2_ref);
-+
         // Capture the stamp from construction to detect updates.
         let original_idle = *idle_at.lock().unwrap();
-+
         let mut a = SingletonConnection::new(
             OkService,
             h2_ref.clone(),
             Arc::new(super::super::stats::ConnectionCounters::default()),
         );
         let mut b = SingletonConnection::new(
             OkService,
             h2_ref.clone(),
             Arc::new(super::super::stats::ConnectionCounters::default()),
         );
         let mut c = SingletonConnection::new(
             OkService,
             h2_ref,
             Arc::new(super::super::stats::ConnectionCounters::default()),
         );
-+
         let _ = a.call(()).await;
         let _ = b.call(()).await;
         let _ = c.call(()).await;
         assert_eq!(active.load(Ordering::Acquire), 3);
-+
         // Sleep so any stamp is distinguishable from construction time.
         tokio::time::sleep(std::time::Duration::from_millis(5)).await;
-+
         // Release two; idle_at must NOT yet be stamped (counter > 0).
         drop(a);
         drop(b);
         assert_eq!(active.load(Ordering::Acquire), 1);
         assert_eq!(
             *idle_at.lock().unwrap(),
             original_idle,
             "idle_at must not be stamped while dispatches are in flight"
         );
-+
         // Last release: 1 → 0 transition. idle_at must be stamped.
         drop(c);
         assert_eq!(active.load(Ordering::Acquire), 0);
         let final_idle = *idle_at.lock().unwrap();
         assert!(
             final_idle > original_idle,
             "idle_at should be stamped when the last DispatchGuard releases"
         );
-+
+    }
-+
+}

Pages: 1 2 3 4 5 6 7 8