1 + | /*
|
2 + | * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
3 + | * SPDX-License-Identifier: Apache-2.0
|
4 + | */
|
5 + |
|
6 + | //! HTTP connection pool.
|
7 + | //!
|
8 + | //! Pool ownership and partition topology are declared at build time:
|
9 + | //!
|
10 + | //! - [`SharedPool`] owns connection lifecycle: TLS, DNS resolution,
|
11 + | //! connection limits, idle eviction, proxy routing, event listening,
|
12 + | //! and the partition registry. Built via [`SharedPool::builder`] which
|
13 + | //! returns a [`Builder`].
|
14 + | //! - [`Client`] is a per-partition view over a [`SharedPool`] that
|
15 + | //! implements [`HttpClient`]. Multiple [`Client`]s may share one
|
16 + | //! [`SharedPool`], each targeting a distinct declared partition.
|
17 + | //!
|
18 + | //! For partition semantics, topologies, and cross-partition checkout
|
19 + | //! policy, see the [`partition`] module.
|
20 + | //!
|
21 + | //! # Example
|
22 + | //!
|
23 + | //! ```no_run
|
24 + | //! # #[cfg(feature = "rustls-aws-lc")]
|
25 + | //! # {
|
26 + | //! use aws_smithy_http_client::pool::{Client, SharedPool};
|
27 + | //! use aws_smithy_http_client::tls;
|
28 + | //! use std::time::Duration;
|
29 + | //!
|
30 + | //! let pool = SharedPool::builder()
|
31 + | //! .tls_provider(tls::Provider::Rustls(
|
32 + | //! tls::rustls_provider::CryptoMode::AwsLc,
|
33 + | //! ))
|
34 + | //! .max_connections(125)
|
35 + | //! .pool_idle_timeout(Duration::from_secs(20))
|
36 + | //! .build_https();
|
37 + | //! let client = Client::new(&pool);
|
38 + | //! # }
|
39 + | //! ```
|
40 + | //!
|
41 + | //! [`HttpClient`]: aws_smithy_runtime_api::client::http::HttpClient
|
42 + |
|
43 + | pub(crate) mod connection;
|
44 + | mod handshake;
|
45 + | pub(crate) mod stats;
|
46 + | mod vendored_cache;
|
47 + |
|
48 + | pub mod builder;
|
49 + | pub mod client;
|
50 + | pub mod partition;
|
51 + |
|
52 + | // Public re-exports.
|
53 + | pub use builder::Builder;
|
54 + | pub use client::Client;
|
55 + | pub use connection::{
|
56 + | Authority, CloseReason, ConnectionClosedEvent, ConnectionCreatedEvent, ConnectionEventListener,
|
57 + | ConnectionFailedEvent, ConnectionReusedEvent, ConnectionTiming, NegotiatedProtocol,
|
58 + | };
|
59 + | pub use partition::{
|
60 + | CrossPartitionPolicy, DriverSpawner, Partition, PartitionId, TokioDriverSpawner,
|
61 + | };
|
62 + | pub use stats::{AuthorityStats, PartitionStats};
|
63 + |
|
64 + | pub(crate) use stats::{ConnectionCounters, StatsIndex};
|
65 + |
|
66 + | /// Connection-caching pool layer.
|
67 + | mod cache {
|
68 + | pub(crate) use super::vendored_cache::*;
|
69 + | }
|
70 + |
|
71 + | use std::collections::HashMap;
|
72 + | use std::convert::Infallible;
|
73 + | use std::future::Future;
|
74 + | use std::pin::Pin;
|
75 + | use std::sync::atomic::{AtomicBool, Ordering};
|
76 + | use std::sync::Arc;
|
77 + | use std::sync::Mutex;
|
78 + | use std::sync::OnceLock;
|
79 + | use std::task::Poll;
|
80 + | use std::time::{Duration, Instant};
|
81 + |
|
82 + | use aws_smithy_runtime_api::box_error::BoxError;
|
83 + | use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
|
84 + | use aws_smithy_types::body::SdkBody;
|
85 + | use hyper_util::client::legacy::connect::Connection as HyperConnection;
|
86 + | use hyper_util::client::pool as hpool;
|
87 + | use hyper_util::client::proxy::matcher::Matcher as ProxyMatcher;
|
88 + | use hyper_util::rt::TokioExecutor;
|
89 + | use tokio::sync::{oneshot, Semaphore};
|
90 + | use tower::{Service, ServiceExt};
|
91 + |
|
92 + | use connection::{
|
93 + | CachedConnection, CheckoutResponse, ConnectionGuard, GuardedBody, H2ConnectionRef,
|
94 + | SingletonConnection,
|
95 + | };
|
96 + | pub(crate) use connection::{ConnectCtx, ReadTimeoutHint, TimeoutContext};
|
97 + | use handshake::{H1ConnectAndHandshake, H1SendRequest, H2ConnectAndHandshake};
|
98 + |
|
99 + | type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send>>;
|
100 + |
|
101 + | /// Request-extension slot the pool checkout fills in with the
|
102 + | /// `ConnectionMetadata` for the selected connection.
|
103 + | ///
|
104 + | /// Read later by the adapter's `CaptureSmithyConnection` retriever, which
|
105 + | /// is what `ConnectionPoisoningInterceptor` uses to decide whether to call
|
106 + | /// `ConnectionMetadata::poison()` on a transient error. Poisoning flips the
|
107 + | /// shared `PoisonPill` on the actual `ManagedConnection`, so the pool
|
108 + | /// skips it on checkout and drops it on return.
|
109 + | ///
|
110 + | /// Write-once per request: `H{1,2}Checkout::call` sets it exactly once
|
111 + | /// during checkout. Subsequent sets are no-ops (the `OnceLock` guarantees
|
112 + | /// single-init). Retries produce fresh `HttpConnector::call` invocations
|
113 + | /// which create fresh capture slots; each attempt's metadata points at
|
114 + | /// the connection used for that attempt.
|
115 + | #[derive(Clone, Default)]
|
116 + | pub(crate) struct ConnectionMetadataCapture {
|
117 + | slot: Arc<std::sync::OnceLock<ConnectionMetadata>>,
|
118 + | }
|
119 + |
|
120 + | impl ConnectionMetadataCapture {
|
121 + | pub(crate) fn new() -> Self {
|
122 + | Self::default()
|
123 + | }
|
124 + |
|
125 + | pub(crate) fn set(&self, metadata: ConnectionMetadata) {
|
126 + | // Silently ignore duplicate sets: single-set is the contract, extra
|
127 + | // sets would only happen via pool-internal bugs.
|
128 + | let _ = self.slot.set(metadata);
|
129 + | }
|
130 + |
|
131 + | pub(crate) fn get(&self) -> Option<ConnectionMetadata> {
|
132 + | self.slot.get().cloned()
|
133 + | }
|
134 + | }
|
135 + |
|
136 + | /// Pool-level configuration.
|
137 + | ///
|
138 + | /// Defaults are applied at the point each setting takes effect rather
|
139 + | /// than in this struct.
|
140 + | #[derive(Clone, Default)]
|
141 + | pub(crate) struct PoolConfig {
|
142 + | /// Upper bound on concurrent connections (total, across all hosts).
|
143 + | /// Enforced via semaphore at the connection establishment layer.
|
144 + | /// `None` = unlimited.
|
145 + | pub(crate) max_connections: Option<usize>,
|
146 + |
|
147 + | /// Upper bound on concurrent connections per host.
|
148 + | /// Each unique (scheme, authority) pair gets an independent semaphore.
|
149 + | /// `None` = unlimited.
|
150 + | pub(crate) max_connections_per_host: Option<usize>,
|
151 + |
|
152 + | /// How long an idle connection may stay in the pool before being
|
153 + | /// evicted. `None` = no eviction.
|
154 + | pub(crate) pool_idle_timeout: Option<std::time::Duration>,
|
155 + |
|
156 + | /// Optional listener for connection lifecycle events.
|
157 + | pub(crate) connection_event_listener: Option<Arc<dyn connection::ConnectionEventListener>>,
|
158 + | }
|
159 + |
|
160 + | /// The connection pool's configuration surface.
|
161 + | ///
|
162 + | /// Owns the connection lifecycle (creation, caching, eviction, health
|
163 + | /// checking) and proxy routing decisions. Multiple [`Client`] instances
|
164 + | /// can reference one `SharedPool`, each presenting a different
|
165 + | /// per-partition view of the same underlying connections.
|
166 + | ///
|
167 + | /// Construct via [`SharedPool::builder`], which returns a [`Builder`].
|
168 + | /// Cloning is cheap (shared via `Arc`).
|
169 + | #[derive(Clone, Debug)]
|
170 + | pub struct SharedPool {
|
171 + | pub(crate) inner: Arc<SharedPoolInner>,
|
172 + | }
|
173 + |
|
174 + | /// Interior of [`SharedPool`]: the connection pool plus the optional proxy
|
175 + | /// matcher consulted per request to decide proxy vs. direct routing.
|
176 + | pub(crate) struct SharedPoolInner {
|
177 + | pub(crate) pool: Arc<ConnectionPool>,
|
178 + | pub(crate) proxy_matcher: Option<Arc<ProxyMatcher>>,
|
179 + | }
|
180 + |
|
181 + | impl std::fmt::Debug for SharedPoolInner {
|
182 + | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
183 + | f.debug_struct("SharedPoolInner").finish_non_exhaustive()
|
184 + | }
|
185 + | }
|
186 + |
|
187 + | impl SharedPool {
|
188 + | /// Create a [`Builder`] for configuring a new connection pool.
|
189 + | pub fn builder() -> Builder<super::TlsUnset> {
|
190 + | Builder::default()
|
191 + | }
|
192 + |
|
193 + | /// Point-in-time, per-partition snapshot of connection counts for `authority`.
|
194 + | ///
|
195 + | /// Relaxed atomics; values may be stale. Sparse — only partitions that have
|
196 + | /// opened a connection to this authority appear.
|
197 + | ///
|
198 + | /// The `authority` must match the form the pool keys on (see
|
199 + | /// [`Authority::from_host`]); a value that matches no keyed authority
|
200 + | /// yields empty stats.
|
201 + | ///
|
202 + | /// [`Authority::from_host`]: crate::client::pool::Authority::from_host
|
203 + | pub fn stats(&self, authority: &Authority) -> stats::AuthorityStats {
|
204 + | self.inner.pool.shared.stats_index().snapshot(authority)
|
205 + | }
|
206 + | }
|
207 + |
|
208 + | /// Key for per-host connection pool routing.
|
209 + | #[derive(Clone, Debug, PartialEq, Eq, Hash)]
|
210 + | pub(crate) struct PoolKey {
|
211 + | scheme: http_1x::uri::Scheme,
|
212 + | authority: http_1x::uri::Authority,
|
213 + | }
|
214 + |
|
215 + | impl PoolKey {
|
216 + | pub(crate) fn from_uri(uri: &http_1x::Uri) -> Option<Self> {
|
217 + | Some(Self {
|
218 + | scheme: uri.scheme()?.clone(),
|
219 + | authority: uri.authority()?.clone(),
|
220 + | })
|
221 + | }
|
222 + | }
|
223 + |
|
224 + | /// Type-erased, single-use handle to a checked-out connection that can
|
225 + | /// dispatch one request.
|
226 + | ///
|
227 + | /// A borrowed peer connection crosses the peer's `dyn PoolEntry` boundary
|
228 + | /// as one of these: the concrete checkout type (`H1Checkout`, holding the
|
229 + | /// peer's `CachedConnection`) carries upstream-unnameable parameters, so
|
230 + | /// it is boxed. Dispatching consumes the handle; the response body holds
|
231 + | /// the underlying connection alive and returns it to *the peer's* pool on
|
232 + | /// drop (it was never the borrower's). Boxed only on the actual-borrow
|
233 + | /// path — the common local path dispatches through the concrete checkout
|
234 + | /// with no erasure.
|
235 + | pub(crate) trait DispatchConn: Send {
|
236 + | /// Liveness check before dispatch. `Err` means the connection is dead;
|
237 + | /// the caller drops the handle and falls back.
|
238 + | fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll<Result<(), BoxError>>;
|
239 + |
|
240 + | /// Dispatch one request, erasing the response body to `SdkBody`.
|
241 + | fn dispatch(
|
242 + | self: Box<Self>,
|
243 + | req: http_1x::Request<SdkBody>,
|
244 + | ) -> BoxFuture<Result<http_1x::Response<SdkBody>, BoxError>>;
|
245 + | }
|
246 + |
|
247 + | /// Type-erased handle to one Negotiate leg's eviction interface.
|
248 + | ///
|
249 + | /// Each `TypedPoolEntry` holds at most two of these (one for its H1 cache,
|
250 + | /// one for its H2 singleton). The two underlying types (`Cache<…>` and
|
251 + | /// `Singleton<…>`) carry upstream-unnameable parameters from `Negotiate`,
|
252 + | /// so they're captured inside `Box<dyn Fn>` closures. The closures are
|
253 + | /// `Fn` (so multiple callers can invoke through `Arc<dyn Fn>`); interior
|
254 + | /// mutability lives in a `std::sync::Mutex` over a clone of the leg, which
|
255 + | /// is uncontended in practice; retain runs at most once per
|
256 + | /// `max(pool_idle_timeout, MIN_EVICTION_TICK)`, and the leg's own state
|
257 + | /// already serializes through its internal `Arc<Mutex<Shared>>`.
|
258 + | pub(crate) struct BoxedRetainer {
|
259 + | retain_fn: Box<dyn Fn(Duration) + Send + Sync + 'static>,
|
260 + | is_empty_fn: Box<dyn Fn() -> bool + Send + Sync + 'static>,
|
261 + | /// Pop one idle connection and drop it to free its permit (active
|
262 + | /// reclaim). Returns `true` if a connection was freed. The H1 cache
|
263 + | /// leg pops from its idle Vec; the H2 singleton leg has no
|
264 + | /// reclaimable idle (an idle H2 connection still multiplexes),
|
265 + | /// so its `reclaim_fn` is a no-op returning `false`.
|
266 + | reclaim_fn: Box<dyn Fn() -> bool + Send + Sync + 'static>,
|
267 + | /// Take one idle connection wrapped as a dispatchable handle that
|
268 + | /// returns to *this* (the owner's) pool on drop — cross-partition
|
269 + | /// borrow. Returns `None` if no idle connection is available. The H1
|
270 + | /// cache leg checks out an idle connection; the H2 singleton leg is a
|
271 + | /// no-op returning `None` (H2 borrow is not supported — an idle H2
|
272 + | /// connection still multiplexes on its owner).
|
273 + | borrow_fn: Box<dyn Fn() -> Option<Box<dyn DispatchConn>> + Send + Sync + 'static>,
|
274 + | }
|
275 + |
|
276 + | impl BoxedRetainer {
|
277 + | fn retain_idle(&self, timeout: Duration) {
|
278 + | (self.retain_fn)(timeout)
|
279 + | }
|
280 + |
|
281 + | fn is_empty(&self) -> bool {
|
282 + | (self.is_empty_fn)()
|
283 + | }
|
284 + |
|
285 + | fn reclaim_one(&self) -> bool {
|
286 + | (self.reclaim_fn)()
|
287 + | }
|
288 + |
|
289 + | fn borrow_one(&self) -> Option<Box<dyn DispatchConn>> {
|
290 + | (self.borrow_fn)()
|
291 + | }
|
292 + | }
|
293 + |
|
294 + | /// Per-host retainer registry. Bounded at two entries (one H1 cache, one
|
295 + | /// H2 singleton); populated lazily by the H1 fallback and H2 upgrade
|
296 + | /// `layer_fn`s the first time `Negotiate` constructs each leg.
|
297 + | type RetainerSlot = Arc<Mutex<Vec<BoxedRetainer>>>;
|
298 + |
|
299 + | /// Per-partition stack factory: builds a host's Negotiate(Cache,Singleton)
|
300 + | /// entry on first touch. Captures the partition's connector; the shared
|
301 + | /// budget/hooks arrive via `&SharedPoolState` at call time.
|
302 + | pub(crate) type MakeStack =
|
303 + | Arc<dyn Fn(&http_1x::Uri, &SharedPoolState) -> Box<dyn PoolEntry> + Send + Sync>;
|
304 + |
|
305 + | /// Which semaphore bound a connect attempt at the cap, identifying what a
|
306 + | /// reclaim must free to relieve it. Acquire order is per-host then global
|
307 + | /// so a per-host failure yields `PerHost`, and a failure on the
|
308 + | /// global semaphore while already holding the per-host permit yields
|
309 + | /// `Global`.
|
310 + | #[derive(Debug, Clone)]
|
311 + | pub(crate) enum BindingConstraint {
|
312 + | /// The global `max_connections` semaphore is exhausted. Any
|
313 + | /// over-supplied peer's idle connection frees a fungible permit.
|
314 + | Global,
|
315 + | /// The per-host `max_connections_per_host` semaphore for this key is
|
316 + | /// exhausted. Only a peer's idle connection *to the same host* frees
|
317 + | /// the relevant permit.
|
318 + | PerHost(PoolKey),
|
319 + | }
|
320 + |
|
321 + | /// Handle for cross-partition active reclaim, held by `ConnectionLimit`.
|
322 + | ///
|
323 + | /// At a cap-bound connect, the requesting partition uses this to free one
|
324 + | /// over-supplied peer's idle connection (dropping it returns its permit to
|
325 + | /// the bound semaphore) before blocking-acquiring. Connection-shaped and
|
326 + | /// NIC-blind: the freed *permit* is what matters; P0 then connects on its
|
327 + | /// own NIC. Candidates are narrowed by the (advisory) stats index and
|
328 + | /// confirmed by the authoritative cache pop.
|
329 + | #[derive(Clone)]
|
330 + | pub(crate) struct PeerReclaimHandle {
|
331 + | /// `Weak` to avoid a cycle: the registry transitively owns the
|
332 + | /// partitions whose `ConnectionLimit`s hold this handle.
|
333 + | registry: std::sync::Weak<partition::PartitionRegistry>,
|
334 + | /// Owned (Arc) so the handle does not borrow `SharedPoolState`.
|
335 + | stats_index: Arc<StatsIndex>,
|
336 + | /// The requesting partition — excluded from its own candidate walk.
|
337 + | self_partition: PartitionId,
|
338 + | }
|
339 + |
|
340 + | impl PeerReclaimHandle {
|
341 + | /// Free one over-supplied peer's idle connection to relieve
|
342 + | /// `constraint`, returning `true` if a permit was freed. Best-effort:
|
343 + | /// `false` if the pool is gone, there are no NIC-group peers, or no
|
344 + | /// peer holds reclaimable idle (P0 then blocks on the permit).
|
345 + | pub(crate) fn try_free_under_load(&self, constraint: &BindingConstraint) -> bool {
|
346 + | let registry = match self.registry.upgrade() {
|
347 + | Some(r) => r,
|
348 + | None => return false, // pool dropped → nothing to reclaim
|
349 + | };
|
350 + | let peers = registry.nic_group_peers(self.self_partition);
|
351 + | if peers.is_empty() {
|
352 + | return false; // alone in the NIC group (e.g. single-partition default)
|
353 + | }
|
354 + | match constraint {
|
355 + | BindingConstraint::PerHost(key) => {
|
356 + | // Candidates = peers with idle to this authority (index
|
357 + | // narrows; cache pop confirms). Round-robin start offset.
|
358 + | let authority = Authority::new(key.authority.as_str());
|
359 + | let mut candidates: Vec<PartitionId> = self
|
360 + | .stats_index
|
361 + | .idle_partitions_for(&authority)
|
362 + | .into_iter()
|
363 + | .map(|(p, _idle)| p)
|
364 + | .filter(|p| *p != self.self_partition && peers.contains(p))
|
365 + | .collect();
|
366 + | self.rotate(&mut candidates);
|
367 + | candidates
|
368 + | .into_iter()
|
369 + | .any(|peer| registry.try_reclaim_on(peer, key))
|
370 + | }
|
371 + | BindingConstraint::Global => {
|
372 + | // Fungible permit: any same-NIC-group peer's idle (any
|
373 + | // authority) relieves the global cap. Narrow to peers that
|
374 + | // the index shows holding idle.
|
375 + | let mut candidates: Vec<PartitionId> = self
|
376 + | .stats_index
|
377 + | .idle_cells()
|
378 + | .into_iter()
|
379 + | .map(|(_authority, p)| p)
|
380 + | .filter(|p| *p != self.self_partition && peers.contains(p))
|
381 + | .collect();
|
382 + | candidates.sort_unstable_by_key(|p| p.as_u64());
|
383 + | candidates.dedup();
|
384 + | self.rotate(&mut candidates);
|
385 + | candidates
|
386 + | .into_iter()
|
387 + | .any(|peer| registry.try_reclaim_any(peer))
|
388 + | }
|
389 + | }
|
390 + | }
|
391 + |
|
392 + | /// Rotate the candidate vec by the partition's advisory `peer_cursor`,
|
393 + | /// so concurrent reclaims from this partition do not all probe the
|
394 + | /// lowest-numbered candidate first. No-op if the registry or partition
|
395 + | /// is gone, or the candidate set is empty.
|
396 + | fn rotate(&self, candidates: &mut [PartitionId]) {
|
397 + | if candidates.len() < 2 {
|
398 + | return;
|
399 + | }
|
400 + | if let Some(registry) = self.registry.upgrade() {
|
401 + | if let Some(state) = registry.partition_opt(self.self_partition) {
|
402 + | let n = candidates.len();
|
403 + | let start = state
|
404 + | .peer_cursor
|
405 + | .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
|
406 + | % n;
|
407 + | candidates.rotate_left(start);
|
408 + | }
|
409 + | }
|
410 + | }
|
411 + | }
|
412 + |
|
413 + | /// Handle for cross-partition borrow (`PreferLocal`), held by the pool
|
414 + | /// entry under a binding cap.
|
415 + | ///
|
416 + | /// At a cap-bound local checkout (`AcquireMode::NonBlocking` returned
|
417 + | /// `CapBound`), the requesting partition uses this to borrow one peer's
|
418 + | /// idle connection and dispatch its request through it — no permit moves,
|
419 + | /// no cold start. Response-shaped and NIC-*bounded*: the borrowed
|
420 + | /// connection physically lives on the peer's NIC, so candidates are drawn
|
421 + | /// only from the same NIC group (unlike reclaim, which frees a fungible
|
422 + | /// permit and is NIC-blind). Always keyed to the requested authority (the
|
423 + | /// borrowed connection must already be connected to that host).
|
424 + | #[derive(Clone)]
|
425 + | pub(crate) struct PeerBorrowHandle {
|
426 + | /// `Weak` to avoid a cycle: the registry transitively owns the
|
427 + | /// partitions whose entries hold this handle.
|
428 + | registry: std::sync::Weak<partition::PartitionRegistry>,
|
429 + | /// Owned (Arc) so the handle does not borrow `SharedPoolState`.
|
430 + | stats_index: Arc<StatsIndex>,
|
431 + | /// The requesting partition — excluded from its own candidate walk.
|
432 + | self_partition: PartitionId,
|
433 + | }
|
434 + |
|
435 + | impl PeerBorrowHandle {
|
436 + | /// Borrow one same-NIC-group peer's idle connection to `key`'s
|
437 + | /// authority, as a dispatchable handle. `None` if the pool is gone,
|
438 + | /// there are no NIC-group peers, or no peer holds idle to this
|
439 + | /// authority (the caller then falls back to a blocking local acquire).
|
440 + | pub(crate) fn try_borrow(&self, key: &PoolKey) -> Option<Box<dyn DispatchConn>> {
|
441 + | let registry = self.registry.upgrade()?;
|
442 + | let peers = registry.nic_group_peers(self.self_partition);
|
443 + | if peers.is_empty() {
|
444 + | return None; // alone in the NIC group (e.g. single-partition default)
|
445 + | }
|
446 + | // Candidates = peers with idle to this authority (index narrows;
|
447 + | // the cache checkout confirms). Round-robin start offset.
|
448 + | let authority = Authority::new(key.authority.as_str());
|
449 + | let mut candidates: Vec<PartitionId> = self
|
450 + | .stats_index
|
451 + | .idle_partitions_for(&authority)
|
452 + | .into_iter()
|
453 + | .map(|(p, _idle)| p)
|
454 + | .filter(|p| *p != self.self_partition && peers.contains(p))
|
455 + | .collect();
|
456 + | self.rotate(&mut candidates);
|
457 + | candidates
|
458 + | .into_iter()
|
459 + | .find_map(|peer| registry.try_borrow_on(peer, key))
|
460 + | }
|
461 + |
|
462 + | /// Rotate the candidate vec by the partition's advisory `peer_cursor`
|
463 + | /// (shared with reclaim), so concurrent borrows from this partition do
|
464 + | /// not all probe the lowest-numbered candidate first.
|
465 + | fn rotate(&self, candidates: &mut [PartitionId]) {
|
466 + | if candidates.len() < 2 {
|
467 + | return;
|
468 + | }
|
469 + | if let Some(registry) = self.registry.upgrade() {
|
470 + | if let Some(state) = registry.partition_opt(self.self_partition) {
|
471 + | let n = candidates.len();
|
472 + | let start = state
|
473 + | .peer_cursor
|
474 + | .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
|
475 + | % n;
|
476 + | candidates.rotate_left(start);
|
477 + | }
|
478 + | }
|
479 + | }
|
480 + | }
|
481 + |
|
482 + | /// Connection-lifecycle machinery shared across every partition: event
|
483 + | /// hooks, the global connection budget, and the per-host budgets. One
|
484 + | /// instance per pool, held behind `Arc` on `ConnectionPool`.
|
485 + | pub(crate) struct SharedPoolState {
|
486 + | pub(crate) hooks: handshake::PoolHooks,
|
487 + | pub(crate) global_sem: Option<Arc<Semaphore>>,
|
488 + | max_connections_per_host: Option<usize>,
|
489 + | per_host_sems: Mutex<HashMap<PoolKey, Arc<Semaphore>>>,
|
490 + | stats_index: Arc<StatsIndex>,
|
491 + | /// Late-bound back-reference to the registry, for cross-partition
|
492 + | /// reclaim. `Weak` (not `Arc`): the registry transitively owns the
|
493 + | /// partitions whose `ConnectionLimit`s reach back through here — a
|
494 + | /// strong reference would be a cycle (pool never drops). Set once,
|
495 + | /// immediately after the registry is built (`build_pool`), because the
|
496 + | /// registry does not exist when `SharedPoolState` is constructed.
|
497 + | /// Upgrade-or-skip: a dropped pool makes reclaim a correct no-op.
|
498 + | registry: OnceLock<std::sync::Weak<partition::PartitionRegistry>>,
|
499 + | }
|
500 + |
|
501 + | impl SharedPoolState {
|
502 + | pub(crate) fn new(config: &PoolConfig) -> Self {
|
503 + | Self {
|
504 + | hooks: handshake::PoolHooks::new(config.connection_event_listener.clone()),
|
505 + | global_sem: config.max_connections.map(|n| Arc::new(Semaphore::new(n))),
|
506 + | max_connections_per_host: config.max_connections_per_host,
|
507 + | per_host_sems: Mutex::new(HashMap::new()),
|
508 + | stats_index: Arc::new(StatsIndex::default()),
|
509 + | registry: OnceLock::new(),
|
510 + | }
|
511 + | }
|
512 + |
|
513 + | /// Bind the registry back-reference. Called once in `build_pool`
|
514 + | /// right after the registry is constructed. Idempotent-safe: a second
|
515 + | /// call is ignored (the first binding wins).
|
516 + | pub(crate) fn set_registry(&self, registry: &Arc<partition::PartitionRegistry>) {
|
517 + | let _ = self.registry.set(Arc::downgrade(registry));
|
518 + | }
|
519 + |
|
520 + | /// A reclaim handle for the given requesting partition, if the
|
521 + | /// registry is bound and still alive. `None` when no registry is set
|
522 + | /// or the pool has been dropped (reclaim is then a no-op).
|
523 + | pub(crate) fn reclaim_handle(&self, self_partition: PartitionId) -> Option<PeerReclaimHandle> {
|
524 + | let registry = self.registry.get()?.clone();
|
525 + | Some(PeerReclaimHandle {
|
526 + | registry,
|
527 + | stats_index: self.stats_index.clone(),
|
528 + | self_partition,
|
529 + | })
|
530 + | }
|
531 + |
|
532 + | /// A borrow handle for the given requesting partition, if the registry
|
533 + | /// is bound and still alive. `None` when no registry is set or the
|
534 + | /// pool has been dropped (borrow is then unavailable; the caller
|
535 + | /// blocks locally).
|
536 + | pub(crate) fn borrow_handle(&self, self_partition: PartitionId) -> Option<PeerBorrowHandle> {
|
537 + | let registry = self.registry.get()?.clone();
|
538 + | Some(PeerBorrowHandle {
|
539 + | registry,
|
540 + | stats_index: self.stats_index.clone(),
|
541 + | self_partition,
|
542 + | })
|
543 + | }
|
544 + |
|
545 + | /// The shared per-host semaphore for `key`, created on first request
|
546 + | /// for that key. Returns `None` when no per-host limit is configured.
|
547 + | /// Shared across all partitions: two partitions to the same host
|
548 + | /// contend on the same budget.
|
549 + | pub(crate) fn per_host_sem(&self, key: &PoolKey) -> Option<Arc<Semaphore>> {
|
550 + | let n = self.max_connections_per_host?;
|
551 + | let mut map = self.per_host_sems.lock().expect("per_host_sems poisoned");
|
552 + | Some(
|
553 + | map.entry(key.clone())
|
554 + | .or_insert_with(|| Arc::new(Semaphore::new(n)))
|
555 + | .clone(),
|
556 + | )
|
557 + | }
|
558 + |
|
559 + | /// Pool-level inverted index for stats reads.
|
560 + | pub(crate) fn stats_index(&self) -> &StatsIndex {
|
561 + | &self.stats_index
|
562 + | }
|
563 + | }
|
564 + |
|
565 + | /// Assemble a [`ConnectionPool`] from a TCP/TLS connector, pool
|
566 + | /// configuration, the declared partitions, and the cross-partition policy.
|
567 + | ///
|
568 + | /// Builds the shared state (budget semaphores, hooks, stats index) and a
|
569 + | /// per-partition stack factory: on first touch of an authority, a partition
|
570 + | /// lazily builds its `Negotiate(ConnectionLimit, Cache/Singleton)` stack,
|
571 + | /// registering that cell's counters into the stats index and binding it to
|
572 + | /// the shared budget. The connector is captured per partition so each can
|
573 + | /// carry its own NIC binding.
|
574 + | pub(crate) fn build_pool<C, IO, F>(
|
575 + | connector_factory: F,
|
576 + | config: PoolConfig,
|
577 + | partitions: Vec<partition::Partition>,
|
578 + | cross_partition_policy: partition::CrossPartitionPolicy,
|
579 + | ) -> ConnectionPool
|
580 + | where
|
581 + | F: Fn(&partition::Partition) -> C + Send + Sync + 'static,
|
582 + | C: Service<http_1x::Uri, Response = IO> + Clone + Send + Sync + 'static,
|
583 + | C::Error: Into<BoxError> + 'static,
|
584 + | C::Future: Unpin + Send + 'static,
|
585 + | IO: hyper::rt::Read + hyper::rt::Write + HyperConnection + Unpin + Send + 'static,
|
586 + | {
|
587 + | let shared = Arc::new(SharedPoolState::new(&config));
|
588 + |
|
589 + | tracing::debug!(
|
590 + | max_connections = ?config.max_connections,
|
591 + | max_connections_per_host = ?config.max_connections_per_host,
|
592 + | pool_idle_timeout = ?config.pool_idle_timeout,
|
593 + | "pool: initialized"
|
594 + | );
|
595 + |
|
596 + | let connector_factory = Arc::new(connector_factory);
|
597 + | let make_stack_for = {
|
598 + | move |partition: &partition::Partition| -> MakeStack {
|
599 + | let connector = connector_factory(partition);
|
600 + | let partition_id = partition.id;
|
601 + | let spawner = partition.spawner.clone();
|
602 + | let cross_partition_policy = cross_partition_policy;
|
603 + | Arc::new(
|
604 + | move |uri: &http_1x::Uri, shared: &SharedPoolState| -> Box<dyn PoolEntry> {
|
605 + | let authority = Authority::new(
|
606 + | uri.authority()
|
607 + | .expect("pool entry URI has authority")
|
608 + | .as_str(),
|
609 + | );
|
610 + |
|
611 + | let counters = Arc::new(ConnectionCounters::default());
|
612 + | shared
|
613 + | .stats_index()
|
614 + | .register(authority.clone(), partition_id, &counters);
|
615 + |
|
616 + | let key = PoolKey::from_uri(uri).expect("pool entry URI has scheme+authority");
|
617 + | let per_host_sem = shared.per_host_sem(&key);
|
618 + | let limited = handshake::ConnectionLimit::new(
|
619 + | connector.clone(),
|
620 + | shared.global_sem.clone(),
|
621 + | per_host_sem,
|
622 + | counters.clone(),
|
623 + | shared.reclaim_handle(partition_id),
|
624 + | );
|
625 + |
|
626 + | let pool_hooks = shared.hooks.clone();
|
627 + |
|
628 + | // Per-host bridge: `H2ConnectAndHandshake` publishes on each new
|
629 + | // handshake; `SingletonConnection` reads the current entry at
|
630 + | // checkout. Clones share the underlying slot.
|
631 + | let h2_ref = H2ConnectionRef::new();
|
632 + |
|
633 + | // Bounded at two: the H1 fallback and H2 upgrade `layer_fn`s
|
634 + | // each push one entry on first construction.
|
635 + | let retainers: RetainerSlot = Arc::new(Mutex::new(Vec::with_capacity(2)));
|
636 + |
|
637 + | let stack =
|
638 + | hpool::negotiate::builder()
|
639 + | .connect(limited)
|
640 + | .inspect(|established: &connection::EstablishedConnection<IO>| {
|
641 + | established.io.connected().is_negotiated_h2()
|
642 + | })
|
643 + | .fallback({
|
644 + | let retainers = retainers.clone();
|
645 + | let pool_hooks = pool_hooks.clone();
|
646 + | let spawner = spawner.clone();
|
647 + | let counters = counters.clone();
|
648 + | tower::layer::layer_fn(move |inspector| {
|
649 + | let cache = cache::builder()
|
650 + | .executor(TokioExecutor::new())
|
651 + | .build(H1ConnectAndHandshake::new(
|
652 + | inspector,
|
653 + | pool_hooks.clone(),
|
654 + | spawner.clone(),
|
655 + | ));
|
656 + | // Capture a clone of the Cache for eviction. `Cache`
|
657 + | // is Clone and shares state via its internal
|
658 + | // `Arc<Mutex<Shared>>`, so the clone here and the
|
659 + | // original-stack consumption below observe the same
|
660 + | // idle set.
|
661 + | let cache_for_retain =
|
662 + | Arc::new(std::sync::Mutex::new(cache.clone()));
|
663 + | let cache_for_empty = cache_for_retain.clone();
|
664 + | let cache_for_reclaim = cache_for_retain.clone();
|
665 + | let cache_for_borrow = cache_for_retain.clone();
|
666 + | retainers.lock().expect("retainer slot poisoned").push(
|
667 + | BoxedRetainer {
|
668 + | retain_fn: Box::new({
|
669 + | let listener = pool_hooks.listener.clone();
|
670 + | move |timeout| {
|
671 + | let now = Instant::now();
|
672 + | cache_for_retain
|
673 + | .lock()
|
674 + | .expect("retain cache lock poisoned")
|
675 + | .retain(|managed| {
|
676 + | let keep = !managed.is_poisoned()
|
677 + | && now.saturating_duration_since(
|
678 + | managed.idle_at(),
|
679 + | ) < timeout;
|
680 + | if !keep {
|
681 + | let reason = if managed.is_poisoned() {
|
682 + | "poisoned"
|
683 + | } else {
|
684 + | "idle_expired"
|
685 + | };
|
686 + | let idle_duration = now
|
687 + | .saturating_duration_since(
|
688 + | managed.idle_at(),
|
689 + | );
|
690 + | tracing::debug!(
|
691 + | conn_id = %managed.conn_id(),
|
692 + | reason,
|
693 + | ?idle_duration,
|
694 + | "pool: connection evicted"
|
695 + | );
|
696 + | if let Some(ref l) = listener {
|
697 + | l.on_closed(&ConnectionClosedEvent::new(
|
698 + | managed.conn_id(),
|
699 + | managed.info.authority.clone(),
|
700 + | managed.info.remote_addr,
|
701 + | if managed.is_poisoned() {
|
702 + | CloseReason::Poisoned
|
703 + | } else {
|
704 + | CloseReason::IdleTimeout
|
705 + | },
|
706 + | None,
|
707 + | ));
|
708 + | }
|
709 + | }
|
710 + | keep
|
711 + | });
|
712 + | }
|
713 + | }),
|
714 + | is_empty_fn: Box::new(move || {
|
715 + | cache_for_empty
|
716 + | .lock()
|
717 + | .expect("is_empty cache lock poisoned")
|
718 + | .is_empty()
|
719 + | }),
|
720 + | reclaim_fn: Box::new({
|
721 + | let listener = pool_hooks.listener.clone();
|
722 + | move || {
|
723 + | // Pop under the cache lock, RELEASE the
|
724 + | // lock, THEN drop the connection (never
|
725 + | // drop while holding the cache Mutex).
|
726 + | let managed = cache_for_reclaim
|
727 + | .lock()
|
728 + | .expect("reclaim cache lock poisoned")
|
729 + | .try_pop_idle();
|
730 + | match managed {
|
731 + | Some(managed) => {
|
732 + | tracing::debug!(
|
733 + | conn_id = %managed.conn_id(),
|
734 + | "pool: connection reclaimed"
|
735 + | );
|
736 + | if let Some(ref l) = listener {
|
737 + | l.on_closed(
|
738 + | &ConnectionClosedEvent::new(
|
739 + | managed.conn_id(),
|
740 + | managed
|
741 + | .info
|
742 + | .authority
|
743 + | .clone(),
|
744 + | managed.info.remote_addr,
|
745 + | CloseReason::Reclaimed,
|
746 + | None,
|
747 + | ),
|
748 + | );
|
749 + | }
|
750 + | // `managed` drops here, after the
|
751 + | // cache lock is released: its
|
752 + | // `ConnectionPermit` returns a
|
753 + | // permit to the shared semaphore.
|
754 + | drop(managed);
|
755 + | true
|
756 + | }
|
757 + | None => false,
|
758 + | }
|
759 + | }
|
760 + | }),
|
761 + | borrow_fn: Box::new({
|
762 + | let listener = pool_hooks.listener.clone();
|
763 + | let counters = counters.clone();
|
764 + | move || {
|
765 + | // Take an idle connection wrapped so it
|
766 + | // returns to THIS (the owner's) cache on
|
767 + | // drop — the borrower dispatches one
|
768 + | // request through it but never owns it.
|
769 + | // `None` if no idle connection is
|
770 + | // available (borrower falls through to
|
771 + | // the next peer or to a blocking local
|
772 + | // acquire).
|
773 + | let cached = cache_for_borrow
|
774 + | .lock()
|
775 + | .expect("borrow cache lock poisoned")
|
776 + | .try_checkout_idle()?;
|
777 + | let checkout = H1Checkout::<()>::new(
|
778 + | CachedConnection::new(
|
779 + | cached,
|
780 + | listener.clone(),
|
781 + | counters.clone(),
|
782 + | ),
|
783 + | );
|
784 + | Some(Box::new(checkout)
|
785 + | as Box<dyn DispatchConn>)
|
786 + | }
|
787 + | }),
|
788 + | },
|
789 + | );
|
790 + | cache.map_response({
|
791 + | let listener = pool_hooks.listener.clone();
|
792 + | let counters = counters.clone();
|
793 + | move |cached| {
|
794 + | H1Checkout::new(CachedConnection::new(
|
795 + | cached,
|
796 + | listener.clone(),
|
797 + | counters.clone(),
|
798 + | ))
|
799 + | }
|
800 + | })
|
801 + | })
|
802 + | })
|
803 + | .upgrade({
|
804 + | let h2_ref = h2_ref.clone();
|
805 + | let retainers = retainers.clone();
|
806 + | let pool_hooks = pool_hooks.clone();
|
807 + | let spawner = spawner.clone();
|
808 + | let counters = counters.clone();
|
809 + | tower::layer::layer_fn(move |inspected| {
|
810 + | let singleton = hpool::singleton::Singleton::new(
|
811 + | H2ConnectAndHandshake::new(
|
812 + | inspected,
|
813 + | h2_ref.clone(),
|
814 + | pool_hooks.clone(),
|
815 + | authority.clone(),
|
816 + | spawner.clone(),
|
817 + | ),
|
818 + | );
|
819 + | let singleton_for_retain =
|
820 + | Arc::new(std::sync::Mutex::new(singleton.clone()));
|
821 + | let singleton_for_empty = singleton_for_retain.clone();
|
822 + | retainers.lock().expect("retainer slot poisoned").push(
|
823 + | BoxedRetainer {
|
824 + | retain_fn: Box::new({
|
825 + | let listener = pool_hooks.listener.clone();
|
826 + | move |timeout| {
|
827 + | let now = Instant::now();
|
828 + | singleton_for_retain
|
829 + | .lock()
|
830 + | .expect("retain singleton lock poisoned")
|
831 + | .retain(|managed| {
|
832 + | // For H2 the connection stays in
|
833 + | // Singleton while serving streams, so
|
834 + | // `idle_at` alone is not a valid
|
835 + | // idleness signal. Keep the
|
836 + | // connection if any stream is in
|
837 + | // flight; only evict when truly idle
|
838 + | // AND `idle_at` has exceeded the
|
839 + | // timeout (stamped on the 1 → 0
|
840 + | // transition by
|
841 + | // `SingletonConnection::drop`).
|
842 + | let keep = !managed.is_poisoned()
|
843 + | && (managed.active_streams_count() > 0
|
844 + | || now.saturating_duration_since(
|
845 + | managed.idle_at(),
|
846 + | ) < timeout);
|
847 + | if !keep {
|
848 + | let reason = if managed.is_poisoned() {
|
849 + | "poisoned"
|
850 + | } else {
|
851 + | "idle_expired"
|
852 + | };
|
853 + | let idle_duration = now
|
854 + | .saturating_duration_since(
|
855 + | managed.idle_at(),
|
856 + | );
|
857 + | tracing::debug!(
|
858 + | conn_id = %managed.conn_id(),
|
859 + | reason,
|
860 + | ?idle_duration,
|
861 + | "pool: connection evicted"
|
862 + | );
|
863 + | if let Some(ref l) = listener {
|
864 + | l.on_closed(&ConnectionClosedEvent::new(
|
865 + | managed.conn_id(),
|
866 + | managed.info.authority.clone(),
|
867 + | managed.info.remote_addr,
|
868 + | if managed.is_poisoned() {
|
869 + | CloseReason::Poisoned
|
870 + | } else {
|
871 + | CloseReason::IdleTimeout
|
872 + | },
|
873 + | None,
|
874 + | ));
|
875 + | }
|
876 + | }
|
877 + | keep
|
878 + | });
|
879 + | }
|
880 + | }),
|
881 + | is_empty_fn: Box::new(move || {
|
882 + | singleton_for_empty
|
883 + | .lock()
|
884 + | .expect("is_empty singleton lock poisoned")
|
885 + | .is_empty()
|
886 + | }),
|
887 + | // An idle H2 connection still multiplexes; it is
|
888 + | // not directly reclaimable like an H1 cache entry.
|
889 + | // Reclaim skips the H2 leg.
|
890 + | reclaim_fn: Box::new(|| false),
|
891 + | // H2 borrow is unsupported: an idle H2
|
892 + | // connection still multiplexes on its owner,
|
893 + | // so there is no idle connection to hand to a
|
894 + | // borrower. Borrow skips the H2 leg.
|
895 + | borrow_fn: Box::new(|| None),
|
896 + | },
|
897 + | );
|
898 + | singleton.map_response({
|
899 + | let h2_ref = h2_ref.clone();
|
900 + | let counters = counters.clone();
|
901 + | move |singled| {
|
902 + | H2Checkout::new(SingletonConnection::new(
|
903 + | singled,
|
904 + | h2_ref.clone(),
|
905 + | counters.clone(),
|
906 + | ))
|
907 + | }
|
908 + | })
|
909 + | })
|
910 + | })
|
911 + | .build();
|
912 + | Box::new(TypedPoolEntry {
|
913 + | stack,
|
914 + | retainers,
|
915 + | counters,
|
916 + | borrow: match cross_partition_policy {
|
917 + | partition::CrossPartitionPolicy::PreferLocal => {
|
918 + | shared.borrow_handle(partition_id)
|
919 + | }
|
920 + | partition::CrossPartitionPolicy::Never => None,
|
921 + | },
|
922 + | })
|
923 + | },
|
924 + | )
|
925 + | }
|
926 + | };
|
927 + |
|
928 + | let partitions = partition::normalize_partitions(partitions, || {
|
929 + | Arc::new(partition::TokioDriverSpawner::current())
|
930 + | });
|
931 + | let registry = Arc::new(partition::PartitionRegistry::build(
|
932 + | partitions,
|
933 + | make_stack_for,
|
934 + | ));
|
935 + |
|
936 + | // Late-bind the registry back-reference for cross-partition reclaim.
|
937 + | // Done here, after the registry exists, because `SharedPoolState` is
|
938 + | // built before it (the `make_stack` closures capture `shared`).
|
939 + | shared.set_registry(®istry);
|
940 + |
|
941 + | ConnectionPool {
|
942 + | config,
|
943 + | shared,
|
944 + | eviction_spawned: AtomicBool::new(false),
|
945 + | drop_notifier: OnceLock::new(),
|
946 + | registry,
|
947 + | }
|
948 + | }
|
949 + |
|
950 + | /// The connection pool.
|
951 + | ///
|
952 + | /// Routes requests by (scheme, authority) to per-host pool stacks.
|
953 + | /// Each host gets a Negotiate stack that selects between HTTP/1.1 (Cache)
|
954 + | /// and HTTP/2 (Singleton) based on ALPN negotiation.
|
955 + | pub(crate) struct ConnectionPool {
|
956 + | /// Pool-wide configuration.
|
957 + | config: PoolConfig,
|
958 + |
|
959 + | /// Shared connection-lifecycle machinery (hooks, semaphores).
|
960 + | shared: Arc<SharedPoolState>,
|
961 + |
|
962 + | /// Immutable partition registry, resolved at build time.
|
963 + | registry: Arc<partition::PartitionRegistry>,
|
964 + |
|
965 + | /// Latches `true` once the eviction task has been spawned. Subsequent
|
966 + | /// `send_request` calls observe the latch and skip the spawn path.
|
967 + | /// Read only when `config.pool_idle_timeout` is set.
|
968 + | eviction_spawned: AtomicBool,
|
969 + |
|
970 + | /// Drop signal for the eviction task. The task awaits the matching
|
971 + | /// `Receiver`; we never send. When `ConnectionPool` drops, this
|
972 + | /// `OnceLock`'s contents drop, the sender drops, the receiver errors
|
973 + | /// with `Canceled`, and the task exits its `select!` loop.
|
974 + | ///
|
975 + | /// `OnceLock` because the task spawns lazily on first `send_request`
|
976 + | /// (at most one task per pool). The task additionally holds a
|
977 + | /// `Weak<ConnectionPool>` as a fallback exit signal.
|
978 + | drop_notifier: OnceLock<oneshot::Sender<Infallible>>,
|
979 + | }
|
980 + |
|
981 + | /// Minimum eviction tick period. Prevents very short `pool_idle_timeout`s
|
982 + | /// from causing the eviction task to spin hot.
|
983 + | const MIN_EVICTION_TICK: Duration = Duration::from_millis(90);
|
984 + |
|
985 + | impl ConnectionPool {
|
986 + | /// Access the immutable partition registry.
|
987 + | pub(crate) fn registry(&self) -> &Arc<partition::PartitionRegistry> {
|
988 + | &self.registry
|
989 + | }
|
990 + |
|
991 + | /// Send a request through the pool.
|
992 + | ///
|
993 + | /// Routes to the appropriate per-host pool stack in the given partition
|
994 + | /// and sends the request. `ctx` carries the routing URI plus
|
995 + | /// per-operation connect-time data (connect_timeout). Per-operation
|
996 + | /// read_timeout, if any, must be attached to `req.extensions_mut()` as
|
997 + | /// [`ReadTimeoutHint`] before calling; the checkout services read it
|
998 + | /// from there.
|
999 + | ///
|
1000 + | /// Takes `self: &Arc<Self>` so the lazy eviction task can hold a
|
1001 + | /// `Weak<Self>` as a fallback exit signal (primary exit is the drop
|
1002 + | /// of `drop_notifier`; the `Weak` insulates the task against a
|
1003 + | /// missed drop signal).
|
1004 + | pub(crate) async fn send_request(
|
1005 + | self: &Arc<Self>,
|
1006 + | partition: &Arc<partition::PartitionState>,
|
1007 + | ctx: ConnectCtx,
|
1008 + | req: http_1x::Request<SdkBody>,
|
1009 + | ) -> Result<http_1x::Response<SdkBody>, BoxError> {
|
1010 + | let key =
|
1011 + | PoolKey::from_uri(&ctx.uri).ok_or("request URI must have scheme and authority")?;
|
1012 + |
|
1013 + | // Lazily spawn the idle-eviction task on first use: no task if the
|
1014 + | // pool is never used, and no task if `pool_idle_timeout` is `None` or zero.
|
1015 + | self.maybe_spawn_eviction_task();
|
1016 + |
|
1017 + | // Dispatch the request through the per-host entry. The lock is
|
1018 + | // held only long enough to look up / create the entry and call
|
1019 + | // `send`; all I/O happens in the returned future, outside the
|
1020 + | // lock.
|
1021 + | let fut = {
|
1022 + | let mut auth = partition.authorities.lock().unwrap();
|
1023 + | if !auth.contains_key(&key) {
|
1024 + | let entry = (partition.make_stack)(&ctx.uri, &self.shared);
|
1025 + | auth.insert(key.clone(), entry);
|
1026 + | }
|
1027 + | auth.get_mut(&key).unwrap().send(ctx, req)
|
1028 + | };
|
1029 + |
|
1030 + | fut.await
|
1031 + | }
|
1032 + |
|
1033 + | /// Spawn the idle-eviction task if it hasn't been spawned yet and
|
1034 + | /// `pool_idle_timeout` is configured.
|
1035 + | ///
|
1036 + | /// Idempotent and cheap after the first call (single relaxed
|
1037 + | /// `AtomicBool` load returns early). No-op without a timeout, with
|
1038 + | /// a zero timeout, or if already spawned.
|
1039 + | fn maybe_spawn_eviction_task(self: &Arc<Self>) {
|
1040 + | let timeout = match self.config.pool_idle_timeout {
|
1041 + | Some(d) if d > Duration::ZERO => d,
|
1042 + | _ => return,
|
1043 + | };
|
1044 + | // Fast path: already spawned.
|
1045 + | if self.eviction_spawned.load(Ordering::Acquire) {
|
1046 + | return;
|
1047 + | }
|
1048 + | // Claim the spawn slot. Only one task per pool ever.
|
1049 + | if self
|
1050 + | .eviction_spawned
|
1051 + | .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
1052 + | .is_err()
|
1053 + | {
|
1054 + | return;
|
1055 + | }
|
1056 + | let (tx, rx) = oneshot::channel::<Infallible>();
|
1057 + | // Invariant: this code path runs exactly once per pool; the
|
1058 + | // `compare_exchange` on `eviction_spawned` above is the unique
|
1059 + | // claim. If `drop_notifier` is already populated, that invariant
|
1060 + | // is broken.
|
1061 + | self.drop_notifier
|
1062 + | .set(tx)
|
1063 + | .expect("drop_notifier set after exclusive spawn-slot claim");
|
1064 + | let weak = Arc::downgrade(self);
|
1065 + | let tick = timeout.max(MIN_EVICTION_TICK);
|
1066 + | tracing::debug!(
|
1067 + | interval = ?tick,
|
1068 + | pool_idle_timeout = ?timeout,
|
1069 + | "pool: eviction task spawned"
|
1070 + | );
|
1071 + | tokio::spawn(eviction_task(weak, rx, timeout));
|
1072 + | }
|
1073 + |
|
1074 + | /// Walk all partitions, dropping idle connections that have exceeded
|
1075 + | /// `timeout` and removing host entries whose retainers are empty.
|
1076 + | ///
|
1077 + | /// Called by the eviction task on each tick. Safe to call when
|
1078 + | /// partitions are empty (no-op).
|
1079 + | fn retain_idle(&self, timeout: Duration) {
|
1080 + | for partition in self.registry.partitions() {
|
1081 + | let partition_id = partition.id;
|
1082 + | let mut evicted: Vec<Authority> = Vec::new();
|
1083 + | {
|
1084 + | let mut auth = partition.authorities.lock().unwrap();
|
1085 + | auth.retain(|key, entry| {
|
1086 + | entry.retain_idle(timeout);
|
1087 + | if entry.is_empty() {
|
1088 + | tracing::debug!("pool: host entry removed (empty after retain)");
|
1089 + | evicted.push(Authority::new(key.authority.as_str()));
|
1090 + | false
|
1091 + | } else {
|
1092 + | true
|
1093 + | }
|
1094 + | });
|
1095 + | // `auth` guard drops here, dropping the removed entries (and
|
1096 + | // the strong `Arc<ConnectionCounters>` they held). Only then
|
1097 + | // can the index prune observe a zero strong count — unless a
|
1098 + | // checkout is still in flight, in which case the cell stays.
|
1099 + | }
|
1100 + | for authority in evicted {
|
1101 + | self.shared
|
1102 + | .stats_index()
|
1103 + | .prune_if_dead(&authority, partition_id);
|
1104 + | }
|
1105 + | }
|
1106 + | }
|
1107 + | }
|
1108 + |
|
1109 + | /// Background loop that drops idle connections past `pool_idle_timeout`.
|
1110 + | ///
|
1111 + | /// On each tick:
|
1112 + | /// 1. If the pool has been dropped (`Weak::upgrade` returns `None`), exit.
|
1113 + | /// 2. Otherwise, walk the host map, drop connections whose `idle_at` is
|
1114 + | /// older than `timeout`, and remove host entries whose retainers
|
1115 + | /// report empty.
|
1116 + | ///
|
1117 + | /// Tick interval is `max(timeout, MIN_EVICTION_TICK)`. The floor exists
|
1118 + | /// so a very short `pool_idle_timeout` doesn't spin the task hot.
|
1119 + | /// Connections live on average between 1× and 2× the tick interval past
|
1120 + | /// last use (sawtooth eviction).
|
1121 + | ///
|
1122 + | /// Exit paths:
|
1123 + | /// - Primary: `drop_notifier` (the `Sender<Infallible>` half) drops when
|
1124 + | /// `ConnectionPool` drops, the receiver errors with `Canceled`, the
|
1125 + | /// `select!` left branch fires, the task exits immediately.
|
1126 + | /// - Secondary: `Weak::upgrade` returns `None`. Belt-and-suspenders for
|
1127 + | /// the unlikely case that the notifier didn't fire; the task will
|
1128 + | /// exit on the next tick.
|
1129 + | async fn eviction_task(
|
1130 + | pool: std::sync::Weak<ConnectionPool>,
|
1131 + | mut drop_notifier: oneshot::Receiver<Infallible>,
|
1132 + | timeout: Duration,
|
1133 + | ) {
|
1134 + | let tick = timeout.max(MIN_EVICTION_TICK);
|
1135 + | let mut interval = tokio::time::interval(tick);
|
1136 + | interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
1137 + | // The first tick of `tokio::time::interval` resolves immediately;
|
1138 + | // burn it so subsequent ticks fire at `now + tick`.
|
1139 + | interval.tick().await;
|
1140 + | loop {
|
1141 + | tokio::select! {
|
1142 + | _ = &mut drop_notifier => break,
|
1143 + | _ = interval.tick() => {
|
1144 + | match pool.upgrade() {
|
1145 + | Some(pool) => pool.retain_idle(timeout),
|
1146 + | None => break,
|
1147 + | }
|
1148 + | }
|
1149 + | }
|
1150 + | }
|
1151 + | tracing::trace!("pool eviction task exiting");
|
1152 + | }
|
1153 + |
|
1154 + | /// Recognize hyper-util's `Singleton` coalescing-cancel sentinel by its
|
1155 + | /// `Display`. The error type (`SingletonError`) and its inner `Canceled`
|
1156 + | /// are upstream-private and not downcastable, so the chain is matched by
|
1157 + | /// string. Deliberately narrow: only the exact "singleton connection
|
1158 + | /// canceled" message re-enters a checkout; any other error propagates.
|
1159 + | /// The string match is exercised by the multi-threaded concurrency test,
|
1160 + | /// which fails if the upstream message changes; if upstream stops
|
1161 + | /// producing the sentinel, this becomes inert.
|
1162 + | fn is_singleton_canceled(err: &(dyn std::error::Error + 'static)) -> bool {
|
1163 + | let mut e = Some(err);
|
1164 + | while let Some(cur) = e {
|
1165 + | if cur.to_string() == "singleton connection canceled" {
|
1166 + | return true;
|
1167 + | }
|
1168 + | e = cur.source();
|
1169 + | }
|
1170 + | false
|
1171 + | }
|
1172 + |
|
1173 + | /// Rewrite an HTTP/1.1 request's URI to the appropriate request-target
|
1174 + | /// form for dispatch:
|
1175 + | ///
|
1176 + | /// - `CONNECT` → authority-form (`host:port`)
|
1177 + | /// - proxied non-CONNECT → absolute-form (full URL with scheme + authority)
|
1178 + | /// - direct non-CONNECT → origin-form (path + query only)
|
1179 + | ///
|
1180 + | /// The HTTP/1 connection sends request URIs as-is, so request-target
|
1181 + | /// form selection is the caller's responsibility.
|
1182 + | fn rewrite_h1_request_target(req: &mut http_1x::Request<SdkBody>, is_proxied: bool) {
|
1183 + | use http_1x::{uri::Parts, Method, Uri};
|
1184 + | if req.method() == Method::CONNECT {
|
1185 + | // Authority-form: retain only the authority component.
|
1186 + | if let Some(auth) = req.uri().authority().cloned() {
|
1187 + | let mut parts = Parts::default();
|
1188 + | parts.authority = Some(auth);
|
1189 + | *req.uri_mut() = Uri::from_parts(parts).expect("authority is valid uri");
|
1190 + | }
|
1191 + | return;
|
1192 + | }
|
1193 + | if is_proxied {
|
1194 + | // Absolute-form: leave the URI as-is (scheme + authority + path).
|
1195 + | return;
|
1196 + | }
|
1197 + | // Origin-form: strip scheme + authority, keep path + query (defaulting
|
1198 + | // to "/" if the URI had nothing after the authority).
|
1199 + | let path_and_query = req
|
1200 + | .uri()
|
1201 + | .path_and_query()
|
1202 + | .filter(|p| p.as_str() != "/")
|
1203 + | .cloned();
|
1204 + | *req.uri_mut() = match path_and_query {
|
1205 + | Some(pq) => {
|
1206 + | let mut parts = Parts::default();
|
1207 + | parts.path_and_query = Some(pq);
|
1208 + | Uri::from_parts(parts).expect("path-and-query is valid uri")
|
1209 + | }
|
1210 + | None => Uri::default(),
|
1211 + | };
|
1212 + | }
|
1213 + |
|
1214 + | /// Tower `Service` wrapper for an H1 pool checkout.
|
1215 + | ///
|
1216 + | /// Produced by the H1 fallback leg (one per checkout from the cache).
|
1217 + | /// Its `Service::call(req)` consumes the held `CachedConnection`, runs
|
1218 + | /// the request, and wraps the response body so the connection returns
|
1219 + | /// to the pool only when the body is fully drained.
|
1220 + | ///
|
1221 + | /// # Single-use
|
1222 + | ///
|
1223 + | /// Each checkout represents one `CachedConnection`. `call` moves the
|
1224 + | /// connection into the response future via `Option::take()`; calling
|
1225 + | /// `call` twice panics (unreachable given our checkout-per-request flow).
|
1226 + | ///
|
1227 + | /// The `UnusedH2Phantom` generic is a phantom type parameter: the H1 path
|
1228 + | /// never populates the H2 variant of `ConnectionGuard` and never holds an
|
1229 + | /// H2 checkout, but both legs must produce the same `CheckoutResponse<…>`
|
1230 + | /// so `Negotiate` can compose them uniformly. The phantom slot here is
|
1231 + | /// whatever type the H2 leg's `SingletonConnection` holds, resolved by
|
1232 + | /// type inference at the pool composition site; this wrapper ignores it.
|
1233 + | pub(crate) struct H1Checkout<UnusedH2Phantom> {
|
1234 + | conn: Option<CachedConnection<H1SendRequest>>,
|
1235 + | _marker: std::marker::PhantomData<fn() -> UnusedH2Phantom>,
|
1236 + | }
|
1237 + |
|
1238 + | impl<UnusedH2Phantom> H1Checkout<UnusedH2Phantom> {
|
1239 + | pub(crate) fn new(conn: CachedConnection<H1SendRequest>) -> Self {
|
1240 + | Self {
|
1241 + | conn: Some(conn),
|
1242 + | _marker: std::marker::PhantomData,
|
1243 + | }
|
1244 + | }
|
1245 + | }
|
1246 + |
|
1247 + | impl<UnusedH2Phantom> Service<http_1x::Request<SdkBody>> for H1Checkout<UnusedH2Phantom> {
|
1248 + | type Response = CheckoutResponse<UnusedH2Phantom>;
|
1249 + | type Error = BoxError;
|
1250 + | type Future = BoxFuture<Result<Self::Response, Self::Error>>;
|
1251 + |
|
1252 + | fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll<Result<(), Self::Error>> {
|
1253 + | self.conn
|
1254 + | .as_mut()
|
1255 + | .expect("H1Checkout::poll_ready after call")
|
1256 + | .poll_ready(cx)
|
1257 + | }
|
1258 + |
|
1259 + | fn call(&mut self, req: http_1x::Request<SdkBody>) -> Self::Future {
|
1260 + | let mut conn = self.conn.take().expect("H1Checkout::call called twice");
|
1261 + | tracing::trace!(conn_id = %conn.conn_id(), uri = %req.uri(), "pool: dispatching request");
|
1262 + | // Populate the adapter-provided capture so the
|
1263 + | // `CaptureSmithyConnection` retriever can return a live
|
1264 + | // `ConnectionMetadata` pointing at THIS connection's poison pill.
|
1265 + | if let Some(capture) = req.extensions().get::<ConnectionMetadataCapture>() {
|
1266 + | capture.set(conn.metadata());
|
1267 + | }
|
1268 + | // Read-timeout hint: bounds request-write + response-headers only.
|
1269 + | let read_timeout = req.extensions().get::<ReadTimeoutHint>().cloned();
|
1270 + | let mut req = req;
|
1271 + | // HTTP/1.1 request-target form depends on whether we're talking to
|
1272 + | // a proxy or directly to the origin. CONNECT always uses
|
1273 + | // authority-form; otherwise proxied = absolute-form, direct =
|
1274 + | // origin-form (path + query only; RFC 7230 §5.3.1).
|
1275 + | rewrite_h1_request_target(&mut req, conn.is_proxied());
|
1276 + | Box::pin(async move {
|
1277 + | let send_fut = conn.call(req);
|
1278 + | let resp = super::timeout::maybe_timeout_future(
|
1279 + | send_fut,
|
1280 + | read_timeout.as_ref().map(|h| h.0.duration),
|
1281 + | read_timeout.as_ref().map(|h| &h.0.sleep_impl),
|
1282 + | super::timeout::TimeoutKind::Read,
|
1283 + | )
|
1284 + | .await?;
|
1285 + | let (parts, body) = resp.into_parts();
|
1286 + | let body = GuardedBody::new(body, ConnectionGuard::H1(conn));
|
1287 + | Ok(http_1x::Response::from_parts(parts, body))
|
1288 + | })
|
1289 + | }
|
1290 + | }
|
1291 + |
|
1292 + | impl<UnusedH2Phantom> DispatchConn for H1Checkout<UnusedH2Phantom>
|
1293 + | where
|
1294 + | UnusedH2Phantom: Send + Sync + 'static,
|
1295 + | {
|
1296 + | fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll<Result<(), BoxError>> {
|
1297 + | Service::poll_ready(self, cx)
|
1298 + | }
|
1299 + |
|
1300 + | fn dispatch(
|
1301 + | mut self: Box<Self>,
|
1302 + | req: http_1x::Request<SdkBody>,
|
1303 + | ) -> BoxFuture<Result<http_1x::Response<SdkBody>, BoxError>> {
|
1304 + | // `H1Checkout::call` produces `CheckoutResponse<…>` whose body is
|
1305 + | // the internal `GuardedBody`; erase it to `SdkBody` at this
|
1306 + | // boundary, exactly as `TypedPoolEntry::send` does for the local
|
1307 + | // path. The guard inside lives on in `SdkBody` until the body is
|
1308 + | // drained, returning the connection to the peer's pool.
|
1309 + | let fut = Service::call(&mut *self, req);
|
1310 + | Box::pin(async move {
|
1311 + | let resp = fut.await?;
|
1312 + | let (parts, body) = resp.into_parts();
|
1313 + | Ok(http_1x::Response::from_parts(
|
1314 + | parts,
|
1315 + | SdkBody::from_body_1_x(body),
|
1316 + | ))
|
1317 + | })
|
1318 + | }
|
1319 + | }
|
1320 + |
|
1321 + | /// Tower `Service` wrapper for an H2 pool checkout.
|
1322 + | ///
|
1323 + | /// Symmetric to `H1Checkout`. The `Singleton` type parameter is the
|
1324 + | /// `SingletonConnection<T>` inner `T`, which at the composition site
|
1325 + | /// resolves to the unnameable `hyper_util::client::pool::singleton::
|
1326 + | /// Singled<ManagedConnection<H2SendRequest>>`. We carry it through
|
1327 + | /// generics and never name it concretely.
|
1328 + | ///
|
1329 + | /// # Single-use
|
1330 + | ///
|
1331 + | /// Same contract as `H1Checkout`.
|
1332 + | pub(crate) struct H2Checkout<Singleton> {
|
1333 + | conn: Option<SingletonConnection<Singleton>>,
|
1334 + | }
|
1335 + |
|
1336 + | impl<Singleton> H2Checkout<Singleton> {
|
1337 + | pub(crate) fn new(conn: SingletonConnection<Singleton>) -> Self {
|
1338 + | Self { conn: Some(conn) }
|
1339 + | }
|
1340 + | }
|
1341 + |
|
1342 + | impl<Singleton> Service<http_1x::Request<SdkBody>> for H2Checkout<Singleton>
|
1343 + | where
|
1344 + | Singleton: Service<http_1x::Request<SdkBody>, Response = http_1x::Response<hyper::body::Incoming>>
|
1345 + | + Send
|
1346 + | + 'static,
|
1347 + | Singleton::Error: Into<BoxError>,
|
1348 + | Singleton::Future: Send + 'static,
|
1349 + | {
|
1350 + | type Response = CheckoutResponse<Singleton>;
|
1351 + | type Error = BoxError;
|
1352 + | type Future = BoxFuture<Result<Self::Response, Self::Error>>;
|
1353 + |
|
1354 + | fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll<Result<(), Self::Error>> {
|
1355 + | self.conn
|
1356 + | .as_mut()
|
1357 + | .expect("H2Checkout::poll_ready after call")
|
1358 + | .poll_ready(cx)
|
1359 + | .map_err(Into::into)
|
1360 + | }
|
1361 + |
|
1362 + | fn call(&mut self, req: http_1x::Request<SdkBody>) -> Self::Future {
|
1363 + | let mut conn = self.conn.take().expect("H2Checkout::call called twice");
|
1364 + | if let Some(id) = conn.metadata().and_then(|m| m.connection_id()) {
|
1365 + | tracing::trace!(conn_id = %id, uri = %req.uri(), "pool: dispatching request");
|
1366 + | }
|
1367 + | // Populate the adapter-provided capture. The metadata is published
|
1368 + | // by `H2ConnectAndHandshake` when a fresh H2 connection is
|
1369 + | // established; if a request somehow reaches us before any metadata
|
1370 + | // is available we leave the capture empty (poison becomes a no-op
|
1371 + | // for that request, matching the capture-absent default).
|
1372 + | if let Some(capture) = req.extensions().get::<ConnectionMetadataCapture>() {
|
1373 + | if let Some(metadata) = conn.metadata() {
|
1374 + | capture.set(metadata);
|
1375 + | }
|
1376 + | }
|
1377 + | // Read-timeout hint: bounds request-write + response-headers only.
|
1378 + | let read_timeout = req.extensions().get::<ReadTimeoutHint>().cloned();
|
1379 + | Box::pin(async move {
|
1380 + | let send_fut = conn.call(req);
|
1381 + | let resp = super::timeout::maybe_timeout_future(
|
1382 + | send_fut,
|
1383 + | read_timeout.as_ref().map(|h| h.0.duration),
|
1384 + | read_timeout.as_ref().map(|h| &h.0.sleep_impl),
|
1385 + | super::timeout::TimeoutKind::Read,
|
1386 + | )
|
1387 + | .await?;
|
1388 + | let (parts, body) = resp.into_parts();
|
1389 + | let body = GuardedBody::new(body, ConnectionGuard::H2(conn));
|
1390 + | Ok(http_1x::Response::from_parts(parts, body))
|
1391 + | })
|
1392 + | }
|
1393 + | }
|
1394 + |
|
1395 + | /// Type-erased per-host pool entry.
|
1396 + | ///
|
1397 + | /// Each host has one of these, wrapping the unnameable Negotiate stack.
|
1398 + | /// Dyn erasure is structural: hosts have different Negotiate compositions
|
1399 + | /// with different unnameable inner types, so the per-partition `authorities`
|
1400 + | /// map cannot hold a single concrete type.
|
1401 + | pub(crate) trait PoolEntry: Send + Sync {
|
1402 + | /// Dispatch a single request through this host's Negotiate stack:
|
1403 + | /// checkout (connecting if needed), post-checkout health filtering,
|
1404 + | /// dispatch, and body-guard setup. The returned response body holds
|
1405 + | /// the connection checked out until it is drained or dropped.
|
1406 + | fn send(
|
1407 + | &mut self,
|
1408 + | ctx: ConnectCtx,
|
1409 + | req: http_1x::Request<SdkBody>,
|
1410 + | ) -> BoxFuture<Result<http_1x::Response<SdkBody>, BoxError>>;
|
1411 + |
|
1412 + | /// Drop idle connections that have exceeded the given timeout, and
|
1413 + | /// drop any connection flagged as poisoned.
|
1414 + | ///
|
1415 + | /// Called by the background eviction task at each tick. Called on the
|
1416 + | /// `Box<dyn PoolEntry>` directly (no `&mut`) because the retainers
|
1417 + | /// carry their own interior mutability (see [`BoxedRetainer`]).
|
1418 + | fn retain_idle(&self, timeout: Duration);
|
1419 + |
|
1420 + | /// Whether this entry has no idle connections (H1 cache empty AND H2
|
1421 + | /// singleton empty). Used to drop empty host entries from the map
|
1422 + | /// after `retain_idle`.
|
1423 + | fn is_empty(&self) -> bool;
|
1424 + |
|
1425 + | /// Pop one idle connection and drop it to free its permit, returning
|
1426 + | /// `true` if one was freed. Drives cross-partition active reclaim: a
|
1427 + | /// starved partition frees an over-supplied peer's idle capacity at
|
1428 + | /// the cap-bound point. The dropped connection's `ConnectionPermit`
|
1429 + | /// releases back to the shared semaphore. Fires `on_closed` with
|
1430 + | /// [`CloseReason::Reclaimed`] for the freed connection. Best-effort:
|
1431 + | /// returns `false` if the entry has no reclaimable idle.
|
1432 + | fn try_reclaim_one(&self) -> bool;
|
1433 + |
|
1434 + | /// Take one idle connection as a dispatchable handle that returns to
|
1435 + | /// this entry's pool on drop, for cross-partition borrow. The borrower
|
1436 + | /// dispatches one request through it; the connection stays this
|
1437 + | /// entry's (no permit moves). Returns `None` if no idle connection is
|
1438 + | /// available. Best-effort — only the H1 cache leg yields a handle (H2
|
1439 + | /// idle still multiplexes on its owner, so the H2 leg returns `None`).
|
1440 + | fn try_borrow_one(&self) -> Option<Box<dyn DispatchConn>>;
|
1441 + | }
|
1442 + |
|
1443 + | /// Concrete PoolEntry wrapping a Negotiate stack.
|
1444 + | ///
|
1445 + | /// `PoolUnnameable` propagates whatever pool-internal unnameable type
|
1446 + | /// flows through the checkout services' `CheckoutResponse`; it's carried
|
1447 + | /// through so `Conn::Response` can name `CheckoutResponse<PoolUnnameable>`.
|
1448 + | struct TypedPoolEntry<S> {
|
1449 + | stack: S,
|
1450 + | /// Retainers for this entry's H1 cache + H2 singleton. Populated
|
1451 + | /// lazily on first use of each leg by the `Negotiate` `layer_fn`s.
|
1452 + | /// Empty until the first request against this host; safe to iterate
|
1453 + | /// at any point (no leg = no-op retain, trivially empty).
|
1454 + | retainers: RetainerSlot,
|
1455 + | // Write handle for this cell's counters; the read path is `StatsIndex`.
|
1456 + | #[allow(dead_code)]
|
1457 + | counters: Arc<ConnectionCounters>,
|
1458 + | /// Cross-partition borrow handle. `Some` only under
|
1459 + | /// `CrossPartitionPolicy::PreferLocal`; `None` under `Never` (and for
|
1460 + | /// the single-partition default, where it would have no NIC-group
|
1461 + | /// peers anyway). Drives the `send` cap-bound borrow branch.
|
1462 + | borrow: Option<PeerBorrowHandle>,
|
1463 + | }
|
1464 + |
|
1465 + | impl<S, Conn, PoolUnnameable> PoolEntry for TypedPoolEntry<S>
|
1466 + | where
|
1467 + | S: Service<ConnectCtx, Response = Conn> + Clone + Send + Sync + 'static,
|
1468 + | S::Error: Into<BoxError> + 'static,
|
1469 + | S::Future: Send + 'static,
|
1470 + | Conn: Service<http_1x::Request<SdkBody>, Response = CheckoutResponse<PoolUnnameable>>
|
1471 + | + Send
|
1472 + | + 'static,
|
1473 + | Conn::Error: Into<BoxError>,
|
1474 + | Conn::Future: Send + 'static,
|
1475 + | PoolUnnameable: Send + Sync + 'static,
|
1476 + | {
|
1477 + | fn send(
|
1478 + | &mut self,
|
1479 + | ctx: ConnectCtx,
|
1480 + | req: http_1x::Request<SdkBody>,
|
1481 + | ) -> BoxFuture<Result<http_1x::Response<SdkBody>, BoxError>> {
|
1482 + | let mut svc = self.stack.clone();
|
1483 + | let borrow = self.borrow.clone();
|
1484 + | Box::pin(async move {
|
1485 + | // Local checkout. The post-checkout `poll_ready` is the
|
1486 + | // reactive health check: the composable pool has no proactive
|
1487 + | // checkout-time health check, so a popped idle connection may
|
1488 + | // be dead (server closed it, driver gone). `poll_ready` Err
|
1489 + | // identifies that; the loop discards it and pops the next,
|
1490 + | // until a live connection is checked out or the connect path
|
1491 + | // is reached. Converges because:
|
1492 + | // - Cache idle set is bounded; each post-checkout `poll_ready`
|
1493 + | // Err flips `is_closed` so `Cached::Drop` skips reinsertion,
|
1494 + | // shrinking the set by one.
|
1495 + | // - Singleton clears to `Empty` on `poll_ready` Err, forcing
|
1496 + | // the next `call` to run a fresh handshake; a handshake
|
1497 + | // failure surfaces as an error from `svc.call` (not the
|
1498 + | // inner `poll_ready`), which we propagate.
|
1499 + | // Under `AcquireMode::NonBlocking`, a cache miss at the connect
|
1500 + | // path returns `CapBound` instead of blocking — surfaced here
|
1501 + | // so the caller can try a peer borrow before committing to wait.
|
1502 + | async fn local_checkout<S, Conn, PoolUnnameable>(
|
1503 + | svc: &mut S,
|
1504 + | ctx: ConnectCtx,
|
1505 + | ) -> Result<Conn, BoxError>
|
1506 + | where
|
1507 + | S: Service<ConnectCtx, Response = Conn>,
|
1508 + | S::Error: Into<BoxError>,
|
1509 + | Conn:
|
1510 + | Service<http_1x::Request<SdkBody>, Response = CheckoutResponse<PoolUnnameable>>,
|
1511 + | Conn::Error: Into<BoxError>,
|
1512 + | {
|
1513 + | loop {
|
1514 + | std::future::poll_fn(|cx| svc.poll_ready(cx))
|
1515 + | .await
|
1516 + | .map_err(Into::into)?;
|
1517 + | let mut checkout = match svc.call(ctx.clone()).await {
|
1518 + | Ok(c) => c,
|
1519 + | Err(e) => {
|
1520 + | let e: BoxError = e.into();
|
1521 + | // A coalesced waiter whose `Singleton` maker
|
1522 + | // bounced to the H1 fallback resolves `Canceled`
|
1523 + | // without being reused or connected. Nothing was
|
1524 + | // established, so re-enter the checkout (the
|
1525 + | // retry becomes a maker or reuses the made
|
1526 + | // connection) rather than fail.
|
1527 + | if is_singleton_canceled(&*e) {
|
1528 + | continue;
|
1529 + | }
|
1530 + | return Err(e);
|
1531 + | }
|
1532 + | };
|
1533 + | if std::future::poll_fn(|cx| checkout.poll_ready(cx))
|
1534 + | .await
|
1535 + | .is_ok()
|
1536 + | {
|
1537 + | return Ok(checkout);
|
1538 + | }
|
1539 + | // drop `checkout` → pool cleanup (H1: discard via
|
1540 + | // CachedConnection::Drop; H2: Singleton already cleared).
|
1541 + | }
|
1542 + | }
|
1543 + |
|
1544 + | // Erase a checkout's `CheckoutResponse` body to `SdkBody` at the
|
1545 + | // `dyn PoolEntry` boundary so consumers never see the internal
|
1546 + | // `GuardedBody<...>`. The guard inside lives on in `SdkBody`
|
1547 + | // until the body drains, returning the connection to its pool.
|
1548 + | fn erase_body<PoolUnnameable>(
|
1549 + | resp: CheckoutResponse<PoolUnnameable>,
|
1550 + | ) -> http_1x::Response<SdkBody>
|
1551 + | where
|
1552 + | PoolUnnameable: Send + Sync + 'static,
|
1553 + | {
|
1554 + | let (parts, body) = resp.into_parts();
|
1555 + | http_1x::Response::from_parts(parts, SdkBody::from_body_1_x(body))
|
1556 + | }
|
1557 + |
|
1558 + | match &borrow {
|
1559 + | // PreferLocal: probe locally without blocking; on a
|
1560 + | // cap-bound miss, borrow a peer's idle connection before
|
1561 + | // falling back to a blocking local acquire.
|
1562 + | Some(handle) => {
|
1563 + | let probe = ctx.clone().with_mode(connection::AcquireMode::NonBlocking);
|
1564 + | match local_checkout(&mut svc, probe).await {
|
1565 + | Ok(mut conn) => {
|
1566 + | let resp = conn.call(req).await.map_err(Into::into)?;
|
1567 + | Ok(erase_body(resp))
|
1568 + | }
|
1569 + | // Cap-bound: the cap is full (no connect was
|
1570 + | // attempted under `NonBlocking`). Borrow a peer's
|
1571 + | // idle connection, else fall back to the
|
1572 + | // authoritative blocking acquire. A genuine connect
|
1573 + | // error (a permit was free but the connect failed)
|
1574 + | // is not `CapBound` and propagates.
|
1575 + | Err(err) if connection::CapBound::is(&*err) => {
|
1576 + | let key = PoolKey::from_uri(&ctx.uri)
|
1577 + | .ok_or("request URI must have scheme and authority")?;
|
1578 + | if let Some(mut borrowed) = handle.try_borrow(&key) {
|
1579 + | // Single-shot liveness check. On death,
|
1580 + | // fall through to the authoritative local
|
1581 + | // acquire rather than walk further peers.
|
1582 + | if std::future::poll_fn(|cx| borrowed.poll_ready(cx))
|
1583 + | .await
|
1584 + | .is_ok()
|
1585 + | {
|
1586 + | // Dispatch through the peer's connection;
|
1587 + | // its driver stays on the peer's runtime,
|
1588 + | // and the connection returns to the peer's
|
1589 + | // pool when the body drains.
|
1590 + | return borrowed.dispatch(req).await;
|
1591 + | }
|
1592 + | // Dead borrowed connection drops here:
|
1593 + | // `CachedConnection::Drop` discards it from
|
1594 + | // the peer's pool and balances the peer's
|
1595 + | // `active`.
|
1596 + | }
|
1597 + | // Borrow miss → authoritative blocking local
|
1598 + | // acquire (blocks on the permit; includes reclaim).
|
1599 + | let mut conn = local_checkout(&mut svc, ctx).await?;
|
1600 + | let resp = conn.call(req).await.map_err(Into::into)?;
|
1601 + | Ok(erase_body(resp))
|
1602 + | }
|
1603 + | Err(err) => Err(err),
|
1604 + | }
|
1605 + | }
|
1606 + | // Never (and single-partition default): blocking local
|
1607 + | // acquire, unchanged.
|
1608 + | None => {
|
1609 + | let mut conn = local_checkout(&mut svc, ctx).await?;
|
1610 + | let resp = conn.call(req).await.map_err(Into::into)?;
|
1611 + | Ok(erase_body(resp))
|
1612 + | }
|
1613 + | }
|
1614 + | })
|
1615 + | }
|
1616 + |
|
1617 + | fn retain_idle(&self, timeout: Duration) {
|
1618 + | let retainers = self.retainers.lock().expect("retainer slot poisoned");
|
1619 + | for r in retainers.iter() {
|
1620 + | r.retain_idle(timeout);
|
1621 + | }
|
1622 + | }
|
1623 + |
|
1624 + | fn is_empty(&self) -> bool {
|
1625 + | let retainers = self.retainers.lock().expect("retainer slot poisoned");
|
1626 + | retainers.iter().all(|r| r.is_empty())
|
1627 + | }
|
1628 + |
|
1629 + | fn try_reclaim_one(&self) -> bool {
|
1630 + | let retainers = self.retainers.lock().expect("retainer slot poisoned");
|
1631 + | // Stop at the first leg that frees a connection. The H1 cache leg
|
1632 + | // pops an idle connection; the H2 leg is a no-op (no
|
1633 + | // reclaimable idle).
|
1634 + | retainers.iter().any(|r| r.reclaim_one())
|
1635 + | }
|
1636 + |
|
1637 + | fn try_borrow_one(&self) -> Option<Box<dyn DispatchConn>> {
|
1638 + | let retainers = self.retainers.lock().expect("retainer slot poisoned");
|
1639 + | // Return the first leg that yields a borrowable handle. The H1
|
1640 + | // cache leg checks out an idle connection; the H2 leg is a no-op
|
1641 + | // returning `None`.
|
1642 + | retainers.iter().find_map(|r| r.borrow_one())
|
1643 + | }
|
1644 + | }
|
1645 + |
|
1646 + | #[cfg(test)]
|
1647 + | mod tests {
|
1648 + | use super::*;
|
1649 + |
|
1650 + | #[test]
|
1651 + | fn test_pool_key() {
|
1652 + | let uri: http_1x::Uri = "http://example.com:8080/path".parse().unwrap();
|
1653 + | let key = PoolKey::from_uri(&uri).unwrap();
|
1654 + | assert_eq!(key.authority.as_str(), "example.com:8080");
|
1655 + | }
|
1656 + |
|
1657 + | #[test]
|
1658 + | fn test_pool_key_missing_scheme() {
|
1659 + | let uri: http_1x::Uri = "/path".parse().unwrap();
|
1660 + | assert!(PoolKey::from_uri(&uri).is_none());
|
1661 + | }
|
1662 + |
|
1663 + | /// A `PoolEntry` that panics if dispatched through and reports itself
|
1664 + | /// empty. Used as a host entry for eviction-lifecycle unit tests where
|
1665 + | /// no request is actually sent.
|
1666 + | struct NullPoolEntry;
|
1667 + | impl PoolEntry for NullPoolEntry {
|
1668 + | fn send(
|
1669 + | &mut self,
|
1670 + | _ctx: ConnectCtx,
|
1671 + | _req: http_1x::Request<SdkBody>,
|
1672 + | ) -> BoxFuture<Result<http_1x::Response<SdkBody>, BoxError>> {
|
1673 + | unreachable!("NullPoolEntry::send called in a test")
|
1674 + | }
|
1675 + | fn retain_idle(&self, _timeout: Duration) {}
|
1676 + | fn is_empty(&self) -> bool {
|
1677 + | true
|
1678 + | }
|
1679 + | fn try_reclaim_one(&self) -> bool {
|
1680 + | false
|
1681 + | }
|
1682 + | fn try_borrow_one(&self) -> Option<Box<dyn DispatchConn>> {
|
1683 + | None
|
1684 + | }
|
1685 + | }
|
1686 + |
|
1687 + | fn pool_with_config(config: PoolConfig) -> Arc<ConnectionPool> {
|
1688 + | let shared = Arc::new(SharedPoolState::new(&config));
|
1689 + | let make_stack_for = |_partition: &partition::Partition| -> MakeStack {
|
1690 + | Arc::new(|_uri, _shared| Box::new(NullPoolEntry) as Box<dyn PoolEntry>)
|
1691 + | };
|
1692 + | let partitions = partition::normalize_partitions(Vec::new(), || {
|
1693 + | Arc::new(partition::TokioDriverSpawner::current())
|
1694 + | });
|
1695 + | let registry = Arc::new(partition::PartitionRegistry::build(
|
1696 + | partitions,
|
1697 + | make_stack_for,
|
1698 + | ));
|
1699 + | Arc::new(ConnectionPool {
|
1700 + | config,
|
1701 + | shared,
|
1702 + | registry,
|
1703 + | eviction_spawned: AtomicBool::new(false),
|
1704 + | drop_notifier: OnceLock::new(),
|
1705 + | })
|
1706 + | }
|
1707 + |
|
1708 + | /// A `make_stack_for` stub that produces `NullPoolEntry` stacks — enough
|
1709 + | /// to exercise registry indexing without standing up real connectors.
|
1710 + | fn null_make_stack_for() -> impl Fn(&partition::Partition) -> MakeStack {
|
1711 + | |_partition| Arc::new(|_uri, _shared| Box::new(NullPoolEntry) as Box<dyn PoolEntry>)
|
1712 + | }
|
1713 + |
|
1714 + | /// The registry's default partition is the first declared.
|
1715 + | #[tokio::test]
|
1716 + | async fn registry_default_partition_is_first_declared() {
|
1717 + | let partitions = vec![
|
1718 + | partition::Partition::new(
|
1719 + | partition::PartitionId::from_index(7),
|
1720 + | partition::TokioDriverSpawner::current(),
|
1721 + | ),
|
1722 + | partition::Partition::new(
|
1723 + | partition::PartitionId::from_index(3),
|
1724 + | partition::TokioDriverSpawner::current(),
|
1725 + | ),
|
1726 + | ];
|
1727 + | let registry = partition::PartitionRegistry::build(partitions, null_make_stack_for());
|
1728 + | assert_eq!(
|
1729 + | registry.default_partition().id,
|
1730 + | partition::PartitionId::from_index(7),
|
1731 + | "first declared partition is the default"
|
1732 + | );
|
1733 + | // Both declared partitions resolve.
|
1734 + | assert!(registry
|
1735 + | .partition_opt(partition::PartitionId::from_index(3))
|
1736 + | .is_some());
|
1737 + | assert!(registry
|
1738 + | .partition_opt(partition::PartitionId::from_index(99))
|
1739 + | .is_none());
|
1740 + | }
|
1741 + |
|
1742 + | /// Declaring the same `PartitionId` twice is a programming error and panics.
|
1743 + | #[tokio::test]
|
1744 + | #[should_panic(expected = "duplicate PartitionId")]
|
1745 + | async fn registry_build_panics_on_duplicate_partition_id() {
|
1746 + | let partitions = vec![
|
1747 + | partition::Partition::new(
|
1748 + | partition::PartitionId::from_index(0),
|
1749 + | partition::TokioDriverSpawner::current(),
|
1750 + | ),
|
1751 + | partition::Partition::new(
|
1752 + | partition::PartitionId::from_index(0),
|
1753 + | partition::TokioDriverSpawner::current(),
|
1754 + | ),
|
1755 + | ];
|
1756 + | let _ = partition::PartitionRegistry::build(partitions, null_make_stack_for());
|
1757 + | }
|
1758 + |
|
1759 + | /// The default (no-topology) path normalizes to a single anonymous
|
1760 + | /// partition that the registry indexes and resolves via `Client::new`.
|
1761 + | #[tokio::test]
|
1762 + | async fn registry_anonymous_default_when_no_partitions_declared() {
|
1763 + | let partitions = partition::normalize_partitions(Vec::new(), || {
|
1764 + | Arc::new(partition::TokioDriverSpawner::current())
|
1765 + | });
|
1766 + | let registry = partition::PartitionRegistry::build(partitions, null_make_stack_for());
|
1767 + | assert_eq!(
|
1768 + | registry.default_partition().id,
|
1769 + | partition::PartitionId::default(),
|
1770 + | "no-topology default is the anonymous partition"
|
1771 + | );
|
1772 + | }
|
1773 + |
|
1774 + | /// Without a `pool_idle_timeout`, `maybe_spawn_eviction_task` is a no-op.
|
1775 + | #[tokio::test]
|
1776 + | async fn eviction_task_no_spawn_without_timeout() {
|
1777 + | let pool = pool_with_config(PoolConfig {
|
1778 + | pool_idle_timeout: None,
|
1779 + | ..PoolConfig::default()
|
1780 + | });
|
1781 + | pool.maybe_spawn_eviction_task();
|
1782 + | assert!(!pool.eviction_spawned.load(Ordering::Acquire));
|
1783 + | assert!(pool.drop_notifier.get().is_none());
|
1784 + | }
|
1785 + |
|
1786 + | /// A zero `pool_idle_timeout` is treated the same as `None`.
|
1787 + | #[tokio::test]
|
1788 + | async fn eviction_task_no_spawn_with_zero_timeout() {
|
1789 + | let pool = pool_with_config(PoolConfig {
|
1790 + | pool_idle_timeout: Some(Duration::ZERO),
|
1791 + | ..PoolConfig::default()
|
1792 + | });
|
1793 + | pool.maybe_spawn_eviction_task();
|
1794 + | assert!(!pool.eviction_spawned.load(Ordering::Acquire));
|
1795 + | }
|
1796 + |
|
1797 + | /// Multiple spawn calls only spawn a single task. The second call
|
1798 + | /// observes `eviction_spawned = true` and returns without touching
|
1799 + | /// `drop_notifier`.
|
1800 + | #[tokio::test]
|
1801 + | async fn eviction_task_spawn_is_idempotent() {
|
1802 + | let pool = pool_with_config(PoolConfig {
|
1803 + | pool_idle_timeout: Some(Duration::from_millis(100)),
|
1804 + | ..PoolConfig::default()
|
1805 + | });
|
1806 + | pool.maybe_spawn_eviction_task();
|
1807 + | assert!(pool.eviction_spawned.load(Ordering::Acquire));
|
1808 + | // drop_notifier gets set once. Grab a pointer to it so we can
|
1809 + | // confirm the second call didn't replace it.
|
1810 + | let first_notifier = pool.drop_notifier.get().expect("notifier set");
|
1811 + | let first_ptr = first_notifier as *const _;
|
1812 + | pool.maybe_spawn_eviction_task();
|
1813 + | let second_notifier = pool.drop_notifier.get().expect("notifier still set");
|
1814 + | let second_ptr = second_notifier as *const _;
|
1815 + | assert_eq!(
|
1816 + | first_ptr, second_ptr,
|
1817 + | "drop_notifier should not be replaced on repeat spawn"
|
1818 + | );
|
1819 + | }
|
1820 + |
|
1821 + | /// `retain_idle` drops host entries whose retainers report empty.
|
1822 + | /// Uses `NullPoolEntry` whose `is_empty` returns `true`, so the first
|
1823 + | /// tick should remove every entry.
|
1824 + | #[tokio::test]
|
1825 + | async fn retain_idle_hosts_drops_empty_entries() {
|
1826 + | let pool = pool_with_config(PoolConfig::default());
|
1827 + | let uri: http_1x::Uri = "http://example.com".parse().unwrap();
|
1828 + | let key = PoolKey::from_uri(&uri).unwrap();
|
1829 + | let partition = pool.registry().default_partition();
|
1830 + | partition
|
1831 + | .authorities
|
1832 + | .lock()
|
1833 + | .unwrap()
|
1834 + | .insert(key.clone(), Box::new(NullPoolEntry));
|
1835 + | assert_eq!(partition.authorities.lock().unwrap().len(), 1);
|
1836 + | pool.retain_idle(Duration::from_secs(30));
|
1837 + | assert_eq!(
|
1838 + | partition.authorities.lock().unwrap().len(),
|
1839 + | 0,
|
1840 + | "empty entry should have been evicted"
|
1841 + | );
|
1842 + | }
|
1843 + |
|
1844 + | #[test]
|
1845 + | fn per_host_sem_shared_by_key() {
|
1846 + | let cfg = PoolConfig {
|
1847 + | max_connections_per_host: Some(2),
|
1848 + | ..PoolConfig::default()
|
1849 + | };
|
1850 + | let s = SharedPoolState::new(&cfg);
|
1851 + | let uri: http_1x::Uri = "https://example.com".parse().unwrap();
|
1852 + | let key = PoolKey::from_uri(&uri).unwrap();
|
1853 + | let a = s.per_host_sem(&key).unwrap();
|
1854 + | let b = s.per_host_sem(&key).unwrap();
|
1855 + | assert!(Arc::ptr_eq(&a, &b), "same key must share one semaphore");
|
1856 + | let uri2: http_1x::Uri = "https://other.com".parse().unwrap();
|
1857 + | let key2 = PoolKey::from_uri(&uri2).unwrap();
|
1858 + | let c = s.per_host_sem(&key2).unwrap();
|
1859 + | assert!(
|
1860 + | !Arc::ptr_eq(&a, &c),
|
1861 + | "different key must get a distinct semaphore"
|
1862 + | );
|
1863 + | }
|
1864 + |
|
1865 + | #[tokio::test]
|
1866 + | async fn two_partitions_independent_storage() {
|
1867 + | let pool = SharedPool::builder()
|
1868 + | .partitions([
|
1869 + | partition::Partition::new(
|
1870 + | partition::PartitionId::from_index(0),
|
1871 + | partition::TokioDriverSpawner::current(),
|
1872 + | ),
|
1873 + | partition::Partition::new(
|
1874 + | partition::PartitionId::from_index(1),
|
1875 + | partition::TokioDriverSpawner::current(),
|
1876 + | ),
|
1877 + | ])
|
1878 + | .build_http();
|
1879 + | let p0 = pool
|
1880 + | .inner
|
1881 + | .pool
|
1882 + | .registry()
|
1883 + | .partition(partition::PartitionId::from_index(0));
|
1884 + | let p1 = pool
|
1885 + | .inner
|
1886 + | .pool
|
1887 + | .registry()
|
1888 + | .partition(partition::PartitionId::from_index(1));
|
1889 + | assert!(
|
1890 + | !Arc::ptr_eq(&p0, &p1),
|
1891 + | "distinct partition ids must yield distinct PartitionState arcs"
|
1892 + | );
|
1893 + | }
|
1894 + |
|
1895 + | /// End-to-end lifecycle: `EstablishingGuard::new` → `promote` →
|
1896 + | /// `ManagedConnection` holds `EstablishedGuard` → drop decrements.
|
1897 + | /// Verifies the wiring from handshake through connection lifetime.
|
1898 + | #[test]
|
1899 + | fn established_set_after_handshake() {
|
1900 + | use connection::{Authority, ConnectionInfo, ConnectionPermit, ManagedConnection};
|
1901 + | use stats::{ConnectionCounters, EstablishingGuard};
|
1902 + |
|
1903 + | let counters = Arc::new(ConnectionCounters::default());
|
1904 + | let authority = Authority::new("example.com:443");
|
1905 + | let partition_id = partition::PartitionId::from_index(0);
|
1906 + |
|
1907 + | // Register in StatsIndex so we can read through that path too
|
1908 + | let index = StatsIndex::default();
|
1909 + | index.register(authority.clone(), partition_id, &counters);
|
1910 + |
|
1911 + | // Simulate: ConnectionLimit creates guard post-permit
|
1912 + | let establishing = EstablishingGuard::new(counters.clone());
|
1913 + | assert_eq!(index.establishing_for(&authority, partition_id), 1);
|
1914 + | assert_eq!(index.established_for(&authority, partition_id), 0);
|
1915 + |
|
1916 + | // Simulate: handshake succeeds, promote
|
1917 + | let established = establishing.promote(stats::PROTO_H1);
|
1918 + | assert_eq!(index.establishing_for(&authority, partition_id), 0);
|
1919 + | assert_eq!(index.established_for(&authority, partition_id), 1);
|
1920 + |
|
1921 + | // Simulate: ManagedConnection holds the EstablishedGuard
|
1922 + | let permit = Arc::new(ConnectionPermit::new(None, None));
|
1923 + | let info = ConnectionInfo {
|
1924 + | remote_addr: None,
|
1925 + | local_addr: None,
|
1926 + | is_proxied: false,
|
1927 + | authority: authority.clone(),
|
1928 + | };
|
1929 + | let conn: ManagedConnection<()> = ManagedConnection::new(
|
1930 + | (),
|
1931 + | info,
|
1932 + | aws_smithy_runtime_api::client::connection::ConnectionId::new(0),
|
1933 + | permit,
|
1934 + | established,
|
1935 + | );
|
1936 + |
|
1937 + | // H2-style: clone shares the same Arc<EstablishedGuard>
|
1938 + | let conn2 = conn.clone();
|
1939 + | assert_eq!(index.established_for(&authority, partition_id), 1);
|
1940 + |
|
1941 + | // Drop one clone — guard still alive via the other
|
1942 + | drop(conn);
|
1943 + | assert_eq!(index.established_for(&authority, partition_id), 1);
|
1944 + |
|
1945 + | // Drop last clone — guard fires, established decrements
|
1946 + | drop(conn2);
|
1947 + | assert_eq!(index.established_for(&authority, partition_id), 0);
|
1948 + | }
|
1949 + | }
|