composable connection pool

[aajtodd]

Summary

Adds an opt-in HTTP client built on hyper-util's composable connection pools, exposed under aws_smithy_http_client::pool. It provides connection-count limits, connection lifecycle events and connection-state queries, and a partition model that binds connections to a driver runtime and an optional network interface while sharing one global connection budget. The default client and the existing Builder are unchanged.

Motivation

• Connection bounding. The pool-based client has no connection-count limit. A process under high concurrency can open unbounded sockets and exhaust file descriptors or overwhelm the peer (SIM P407972172). This adds max_connections and max_connections_per_host as connection-level caps — distinct from request concurrency, which H2 multiplexing decouples from connection count.
• hyper-util composable pools. hyper-util's legacy::Client is positioned for eventual removal. Its replacement is a set of composable pool layers — cache, singleton, negotiate, map (overview). This builds the client on those layers, which is also where connection-count limiting becomes expressible (a tower concurrency limit at the connection-establishment layer rather than a request gate).
• Connection-state visibility and locality. A single shared pool can hold a cross-connection view and bind connections to a runtime and interface. N independent pools cannot: each sees only its own connections, and a per-pool connection cap fragments the global budget. The lifecycle events, the connection-state read API, and the partition model below all follow from holding that view in one place.

Scope and compatibility

• Additive. New public surface under aws_smithy_http_client::pool; one additive field (ConnectionId) on aws-smithy-runtime-api's ConnectionMetadata.
• The default HTTP client selection is unchanged. No behavior version gates on this. Existing Builder::new() and its connectors are untouched.
• Opt-in: a consumer reaches it through pool::SharedPool::builder().

Approach

The existing client's observable behavior was captured as a test suite before the pool was written, and the pool was built against it. The suite is therefore both a parity gate — the pool must not regress connection reuse, idle eviction, timeout, poisoning, or proxy behavior — and the specification the implementation targets. The net-new partition surface is tested on top of that baseline. This is why roughly a third of the diff is tests, and why the wire harness (below) is itself substantial: the behavior under test lives at the socket, so the harness had to come first.

Model

A pool serves many authorities; per authority, a request flows through a stack of hyper-util layers. Reuse short-circuits before the connection limit and the handshake; only new connections take the full path.

Connector stack (per authority)

request
  │
  ▼
Map ───────────────── route by (scheme, authority)
  └─ Negotiate ────── select leg by ALPN result
       ├─ H1: Cache ─────────── reuse idle, else ↓
       └─ H2: Singleton ─────── reuse shared conn, else ↓
                          │
                   ConnectionLimit ──── acquire per-host then global permit
                          │              (reuse bypasses this entirely)
                          ▼
                     handshake ──────── driver task spawned on the
                          │             partition's runtime
                          ▼
                        TLS ─────────── TCP

Ownership: per-partition vs shared

Connections never move: a connection is a driver task pinned to the runtime that created it plus a socket bound to that runtime's reactor and a NIC. So storage is per-partition. The budget and the connection-state index are shared, which is what makes one global cap across partitions possible.

SharedPool                       cloneable handle
└─ ConnectionPool
   ├─ shared state ............. global_sem · per_host_sems ·
   │                             connection-state index · event hooks
   ├─ registry (immutable) ..... partitions by id, partitions by NIC group
   └─ partitions
        └─ PartitionState ...... driver spawner · optional NIC
             └─ per-authority storage
                  └─ Negotiate(Cache | Singleton) + connection counters

Client ──▶ one PartitionState    resolved once at construction;
                                 the reuse hot path takes no global lock

The single-partition default is this same structure with one anonymous partition: no NIC groups, no peers, the cap is just that partition's.

Public API and use cases

Path: aws_smithy_http_client::pool::*. The configuration surface is SharedPool (built via SharedPool::builder()); Client is a lightweight handle that implements HttpClient.

1. Bound connection use

max_connections caps total connections; max_connections_per_host caps per authority. Acquired only when establishing a connection — reuse and H2 multiplexing do not consume the budget. Per-host is acquired before global, so a saturated authority blocks only its own connects.

let pool = SharedPool::builder()
    .tls_provider(tls::Provider::Rustls(CryptoMode::AwsLc))
    .max_connections(125)
    .max_connections_per_host(10)
    .build_https();
let client = Client::new(&pool);

Prevents a process from exhausting sockets or file descriptors under high concurrency. See examples/pool-basic.rs.

2. Observe connections

A ConnectionEventListener receives created / reused / closed / failed events carrying a stable ConnectionId, the authority, the negotiated protocol, a close reason, and connect timing. stats(&authority) returns a point-in-time, per-partition read of connection counts. Both are read-only; neither couples the pool to a metrics crate.

struct Metrics;
impl ConnectionEventListener for Metrics {
    fn on_created(&self, event: &ConnectionCreatedEvent) {
        record(event.conn_id(), event.authority(), event.protocol());
    }
    fn on_closed(&self, event: &ConnectionClosedEvent) {
        record_close(event.authority(), event.reason());
    }
}

let pool = SharedPool::builder()
    .tls_provider(tls::Provider::Rustls(CryptoMode::AwsLc))
    .connection_event_listener(Arc::new(Metrics))
    .build_https();

let stats = pool.stats(&Authority::from_host("example.com"));
for (partition, s) in stats.iter() {
    // established: connections that exist; idle(): those not in use.
    report(partition, s.established, s.idle());
}

Surfaces connection churn and cache-hit behavior, and attributes failures to a connection.

3. Keep I/O local on topology-aware runtimes

A partition is a group of connections that share a driver runtime and an optional network interface. On a current-thread-per-core runtime, a connection's driver task is pinned to the runtime that created it; using a connection from another runtime costs a cross-thread wakeup per request. One partition per runtime keeps each connection's I/O on its owning runtime, while a single shared pool still enforces one global connection budget — which N independent per-runtime pools cannot.

Partition::interface(nic) binds that partition's sockets to a network interface (SO_BINDTODEVICE on Linux), so each partition's connections egress its declared NIC. A partition builds its own connector bound to its NIC; the binding is also what defines the NIC groups borrow and reclaim respect (use case 4).

   core 0 runtime          core 1 runtime          core 2 runtime
   partition 0             partition 1             partition 2
   ├─ connections          ├─ connections          ├─ connections
   │  (drivers pinned)     │                        │
   └───────────────┬───────┴────────────┬───────────┘
                   one shared connection budget

let pool = SharedPool::builder()
    .tls_provider(tls::Provider::Rustls(CryptoMode::AwsLc))
    .max_connections(2_000)
    .partitions((0..n).map(|i| {
        Partition::new(PartitionId::from_index(i), spawner_for(i))
    }))
    .build_https();
let client = Client::from_partition(&pool, PartitionId::from_index(i));

See examples/pool-partitioned.rs.

4. Cross-partition behavior under a cap

When a partition has no local idle connection and the cap binds, CrossPartitionPolicy governs what happens. Both behaviors fire only under a binding cap and only within a NIC group — borrowing or reclaiming across a NIC is physically wrong, since the peer's connection is on the wrong interface.

let pool = SharedPool::builder()
    .tls_provider(tls::Provider::Rustls(CryptoMode::AwsLc))
    .max_connections(2_000)
    .cross_partition_policy(CrossPartitionPolicy::PreferLocal)
    .partitions([
        Partition::new(PartitionId::from_index(0), spawner_for(0)).interface("eth0"),
        Partition::new(PartitionId::from_index(1), spawner_for(1)).interface("eth1"),
    ])
    .build_https();

• Never (default) — strict locality. The partition reclaims a peer's permit (then connects locally) or waits; it always serves the request on its own connection.
• PreferLocal — borrow. The partition dispatches the request through a same-NIC peer's existing connection to bridge a transient burst, with no handshake and no permit.

Cap-bound decision (synchronous, per request)

A single decision per request, rooted at the permit acquire:

                      request, local cache miss
                                │
                        try_acquire permit
                                │
              ┌─────────────────┴──────────────────┐
          acquired                            NoPermits  (cap bound)
              │                                     │
        connect locally           binding constraint = PerHost | Global
      (no cross-partition path)                     │
                            ┌─────────────────┬──────┴─────────────┐
                       policy Never                      policy PreferLocal
                            │                                     │
                         RECLAIM                               BORROW
              candidates: same-NIC peers          candidates: same-NIC peers
              the index shows holding idle         the index shows holding idle
              (rotate by peer_cursor)              (rotate by peer_cursor)
                            │                                     │
              first peer whose idle pops+         first peer whose idle pops:
              drops frees a permit:               dispatch request through the
                PerHost → free idle to that       PEER's connection (its driver
                  authority                       stays on the peer's runtime)
                Global → free any peer's idle                    │
                  (permit is fungible)            no peer idle ──▶ falls back to
                            │                       the Never blocking acquire
              blocking-acquire freed permit
                ──▶ connect LOCAL
                            │
              none reclaimable ──▶ block;
              eviction is the fallback ──────────▶ (see eviction diagram)

The index only narrows candidates (advisory); the cache pop is the authoritative gate. A stale index entry can only shrink the candidate set — never cause a wrong action.

Cap-bound fallback (asynchronous, eviction tick)

When Never finds no reclaimable peer idle, the starved partition unblocks through the shared semaphore on the next eviction tick — not through a shared eviction view:

a single eviction task (one per pool) walks ALL partitions each tick:
  for each partition → each authority → retain idle
                                        (drop expired / poisoned)

  P1's idle connection expires ─▶ dropped ─▶ its ConnectionPermit returns
                                                 │
                                          global_sem += 1
                                                 │
        P0 blocked on global_sem (on another runtime) WAKES ─▶ connects LOCAL

Active reclaim does this synchronously at the cap point; eviction does it on the tick. The unblock rides the shared semaphore either way, so it works regardless of which partition evicts.

Testing

The behavior under test sits at the socket boundary — reuse, idle eviction, stale-connection detection, cap pressure, cross-partition borrow and reclaim are only observable there — so the transport is not mocked. Tests run against a purpose-built loopback wire harness: real TCP bound on distinct loopback IPs, per-connection programmable behavior (respond / reset / hold / idle-close), and a server-side event log of what each endpoint saw. The harness (src/test_util/wire/connection.rs) is the lens for the whole suite and is itself substantial net-new infrastructure.

By behavior class, with where each lives:

• Pool behavior (parity) — reuse, idle eviction, stale connection detected at checkout, poisoning (a poisoned connection is not reused), connect and read timeouts, max_connections enforcement, and per-host cap isolation. tests/pool_behavior_test.rs.
• Protocol (H2) — multiplexing on one connection, GOAWAY handling, poisoning, ALPN leg selection. tests/h2_pool_test.rs.
• Topology (partitions) — per-partition spawner isolation, per-partition storage isolation, cross-partition reclaim (Never), cross-partition borrow (PreferLocal, asserting the request ran on the peer's exact connection), and the NIC-group boundary. tests/pool_behavior_test.rs.
• Connection state — counters track the create / reuse / evict lifecycle; stats() reads are sparse (only touched partitions appear), reflect an in-flight checkout, and prune after eviction. tests/pool_behavior_test.rs.
• Concurrency — a multi-partition stress test driving concurrent borrow / reclaim / eviction, run under ThreadSanitizer in the per-crate CI hook (additional-ci). The connection-state counters are advisory and relaxed — drift-tolerant, but not race-tolerant (a relaxed-atomic data race is still undefined behavior), which is what TSan guards. tests/pool_behavior_test.rs.
• TLS and proxy (parity) — provider matrix and proxy routing, run for the pool-based client alongside the existing client. tests/tls.rs, tests/proxy_tests.rs.
• Cross-stack — the orchestrator reconnect-on-transient-error suite runs against the pool-based client alongside the existing hyper stacks. aws-smithy-runtime/tests/reconnect_on_transient_error.rs.

How to review

The review follows a request through the pool, then covers the cross-cutting concerns a single request does not touch. The send path and the connection guards are where correctness concentrates.

The Model section above is the mental model; read it first. Sizes: ~12k insertions across 31 files; ~57% pool source, ~36% tests, the remainder vendored cache, TLS wiring, runtime-api, and build.

Tracing a request

The request enters. Client resolved its partition at construction, so the request goes straight to that partition's authority map. src/client/pool/client.rs, src/client/pool.rs (send_request).

Notes:

• The partition is resolved once, at Client construction, not per request. The reuse hot path takes no global lock — storage is per-partition, and only the cap and the connection-state index are shared.

Reuse, or decide to connect. The Negotiate stack tries an idle H1 connection or the shared H2 connection before anything else; only a miss proceeds toward a new connection. src/client/pool.rs (the Negotiate assembly).

Notes:

• Reuse short-circuits above the connection limit and the handshake — a cache hit acquires no permit and runs no handshake. The permit and handshake are only on the new-connection path.

The permit (the cap). A new connection acquires the per-host permit, then the global permit. src/client/pool/handshake.rs (ConnectionLimit).

Notes:

• Per-host is acquired before global, deliberately: the reverse order lets a saturated authority hold global permits while waiting on its own, which would stall connects to unrelated authorities.
• A failed try_acquire here is the entry to the cross-partition path (covered under Cross-cutting concerns).

Handshake and the driver. The connection is established and its driver task is spawned on the partition's runtime, not the ambient one that issued the request. The managed connection and its RAII guards are created here. src/client/pool/handshake.rs, src/client/pool/connection.rs.

Notes:

• The guards are the subtle part. established is decremented exactly once across the N H2 clones of a connection — it rides the shared inner, not each clone, so dropping a multiplexed-stream handle does not under-count.
• The establishing guard uses promote-vs-drop: a connect that is cancelled before it completes drops the guard and decrements, so a cancelled handshake does not leak a permanent "warming" count.

Response, then return. The connection returns to the pool only when the response body guard drops. src/client/pool/connection.rs (the body guard).

Notes:

• The connection stays checked out until the body is fully consumed (the guard rides the response body), so a half-read body does not return a connection that is still in use.
• H1 holds the connection for the single stream; H2 releases per-stream while the connection itself stays multiplexed.

Cross-cutting concerns

Connection identity and events. A stable ConnectionId and the lifecycle listener (created / reused / closed / failed). aws-smithy-runtime-api/src/client/connection.rs (the additive ConnectionId — semver-relevant public surface), the event types in src/client/pool/connection.rs.

Notes:

• The runtime-api change is purely additive: a new field on ConnectionMetadata and its builder, no change to existing signatures.

Connection state. The counters and the per-(partition, authority) index behind stats(). src/client/pool/stats.rs.

Notes:

• The index is advisory and relaxed by design. A stale read costs a missed routing optimization, never correctness — a request dispatched on a stale hint still succeeds. This is what lets the read path stay lock-free.

Cross-partition behavior. The cap-bound decision (the two diagrams above), the reclaim and borrow handles, and the policy. src/client/pool.rs (the cap-bound branch and the handles), src/client/pool/partition.rs (CrossPartitionPolicy, NIC grouping).

Notes:

• Candidates are NIC-scoped: borrow and reclaim only consider same-NIC-group peers, because a peer's connection physically lives on the peer's interface.
• The connection-state index only narrows the candidate set; the cache pop is the authoritative gate. A stale index entry can shrink the candidate set but cannot cause a wrong action.

Supporting material

Lower scrutiny, in support of the above:

• src/client/pool/vendored_cache.rs — vendored from hyper-util with two additions marked // SDK MODIFICATION; see NOTICE.
• src/client/pool/builder.rs — each partition builds its own connector via a Fn(&Partition) -> Connector factory; bind_interface is the single seam that applies set_interface (NIC binding), guarded to Linux-like targets. The factories differ only in their TLS/proxy wrap.
• src/client/tls/{rustls,s2n_tls}_provider.rs — connector wrapping for connect timing.
• src/client.rs, src/client/timeout.rs, src/client/proxy.rs — visibility changes, shared proxy-auth helper, timeout helper.
• tests/* — grouped by the behavior classes in Testing above; read the harness (src/test_util/wire/connection.rs) first, as it is the lens for every test.

Not in this PR

Deferred, tracked separately:

• DNS resolver feedback (caching, shuffling, failed-IP deprioritization).
• ECONNRESET first-read detection and zero-delay retry.
• Switching the default HTTP client to this implementation behind a behavior version.
• Per-phase connect timing (TCP vs TLS) on connection events.

[github-actions]

A new generated diff is ready to view.

• AWS SDK (ignoring whitespace)
• Client Test (ignoring whitespace)
• No codegen difference in the Server Test
• No codegen difference in the Server Test Python
• No codegen difference in the Server Test Typescript

A new doc preview is ready to view.

[landonxjames]

Quick thought (just reading the description now, haven't dived into the code yet), but do we want to focus on the pool here? I know that is our main motivator for shipping this, but I think hyper wants to include more of this composable functionality in the future that isn't just focus on the pool. Maybe composable would be a better module name?

Maybe it comes down to whether we envision having multiple clients that are focused on different use cases, or just one more configurable/composable client that we expand on in the future.

[ysaito1001]

Reviewed primarily around

• static ownership relationships between key players (e.g. SharedPool, ConnectionPool, SharedPoolState, PartitionRegistry, PartitionState, TypedPoolEntry, ConnectionLimit)
• runtime behaviors around
  • establishing a new connection
  • connection cache hit
  • per-host connection max reached
  • global connection max reached

Looks fantastic. Could be review fallout, but the change carries lots of value as-is.

[ysaito1001]

Modifications can be useful for hyper_util users in general. Plan to upstream these to hyper-util and remove the vendored copy at some point?

[ysaito1001]

Looks like it's a bare invocation of tokio::spawn in a non-feature-gated place like rt-tokio, but I suspect this place assumes async runtime being used is tokio by virtue of using hyper-util?

[landonxjames]

Approved. One maybe bug with H1 connections you should probably look at before merging, other things are mostly just questions.

Overall feedback, this is a ton of code to own for what we get out of it. Agree with Yuki that we should look into upstreaming whatever parts of this we can.

[landonxjames]

I think an H1 host entry can be evicted while a request is still in flight. Meaning a healthy
connection is dropped instead of reused.

retain_idle (line 1074) removes a host entry when is_empty() is true, and the H1 retainer's is_empty is Cache::is_empty() -> shared.services.is_empty(). A checked-out H1 connection has been take() out of services, so for a host whose only connection is mid-request, services is empty -> the entry is removed and the Cache (Arc<Mutex<Shared>>) drops. The in-flight checkout holds only a Weak, so when the body finishes, Cached::Drop's shared.upgrade() returns None and the connection is dropped rather than returned.

H2 doesn't seem to have the same issue (the connection stays in Singleton).

I think this could be fixed by adding a check to this function like:

fn is_empty(&self) -> bool {
    // an entry with in-flight checkouts (`active`) or in-progress connects
    // (`establishing`) must not be removed: a checked-out H1 connection is
    // not in the cache idle set, so the retainers report empty while it is
    // still out. Removing the entry drops the cache it would return to.
    if self.counters.active.load(Ordering::Relaxed) > 0
        || self.counters.establishing.load(Ordering::Relaxed) > 0
    {
        return false;
    }
    let retainers = self.retainers.lock().expect("retainer slot poisoned");
    retainers.iter().all(|r| r.is_empty())
}

[landonxjames]

Quick tests for this added in rust-runtime/aws-smithy-http-client/tests/pool_behavior_test.rs

/// Repro: hold an H1 response (body undrained, connection checked out, so the
/// H1 cache's idle set is empty) across an eviction tick, then drain it and
/// issue a second request to the same host. If the host entry is removed while
/// the request is in flight, the checked-out connection cannot return to its
/// (dropped) cache, so request 2 must reconnect (tcp_accepted == 2).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn review_repro_h1_entry_eviction_during_in_flight_request() {
    use http_body_util::BodyExt;

    let harness = ConnectionTestHarness::builder()
        .endpoint(
            IP1,
            vec![
                ConnectionBehavior::RespondKeepAlive { status: 200, body: b"one" },
                ConnectionBehavior::RespondKeepAlive { status: 200, body: b"two" },
            ],
        )
        .build()
        .await;

    let idle_timeout = Duration::from_millis(100);
    let pool = SharedPool::builder()
        .dns_resolver(harness.dns_resolver())
        .pool_idle_timeout(idle_timeout)
        .build_http();

    let port = harness.endpoints[0].port();
    let url = format!("http://127.0.0.1:{port}/");
    let client = PoolClient::new(&pool).into_shared();

    // Request 1: hold the response without draining, connection checked out.
    let resp1 = send_to(&client, &url).await.expect("req1 should succeed");
    assert_eq!(resp1.status().as_u16(), 200);

    // Let the eviction task tick (>= 2 ticks) while the request is in flight.
    tokio::time::sleep(idle_timeout * 3).await;

    // Drain req1's body: the body guard drops, CachedConnection::Drop fires.
    let _ = BodyExt::collect(resp1.into_body()).await.expect("body1 readable").to_bytes();
    tokio::task::yield_now().await;

    // Request 2 to the same host.
    let resp2 = send_to(&client, &url).await.expect("req2 should succeed");
    assert_eq!(resp2.status().as_u16(), 200);
    let _ = BodyExt::collect(resp2.into_body()).await.expect("body2 readable").to_bytes();

    let accepts = harness.tcp_accepted_count();
    eprintln!("[review-repro] tcp_accepted_count = {accepts}");
    assert_eq!(accepts, 2, "Finding #1: entry evicted in-flight -> req2 reconnects");
}

/// Control: same setup, no eviction tick (60s idle timeout) -> req2 reuses (1 accept).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn review_control_h1_in_flight_no_eviction_reuses() {
    use http_body_util::BodyExt;

    let harness = ConnectionTestHarness::builder()
        .endpoint(
            IP1,
            vec![
                ConnectionBehavior::RespondKeepAlive { status: 200, body: b"one" },
                ConnectionBehavior::RespondKeepAlive { status: 200, body: b"two" },
            ],
        )
        .build()
        .await;

    let pool = SharedPool::builder()
        .dns_resolver(harness.dns_resolver())
        .pool_idle_timeout(Duration::from_secs(60))
        .build_http();

    let port = harness.endpoints[0].port();
    let url = format!("http://127.0.0.1:{port}/");
    let client = PoolClient::new(&pool).into_shared();

    let resp1 = send_to(&client, &url).await.expect("req1 should succeed");
    assert_eq!(resp1.status().as_u16(), 200);
    let _ = BodyExt::collect(resp1.into_body()).await.expect("body1 readable").to_bytes();
    tokio::task::yield_now().await;

    let resp2 = send_to(&client, &url).await.expect("req2 should succeed");
    assert_eq!(resp2.status().as_u16(), 200);
    let _ = BodyExt::collect(resp2.into_body()).await.expect("body2 readable").to_bytes();

    let accepts = harness.tcp_accepted_count();
    eprintln!("[review-control] tcp_accepted_count = {accepts}");
    assert_eq!(accepts, 1, "control: no eviction -> req2 reuses the connection");
}

[aajtodd]

Nice catch, added tests and fix.

[landonxjames]

Think there should be an on_closed() here?

[aajtodd]

There is below.

[landonxjames]

Probably worth a tracing::warn here (or maybe even some kind of failure) for users on unsupported platforms (Mac/Windows) who set a nic. The silent drop seems like surprising behavior.

[aajtodd]

Was missing the cfg guards at the configuration level, now you can only set a nic on linux which was the intent and how current configuration has it.

[landonxjames]

Is it worth confirming the requirement at runtime?

[aajtodd]

Logged at build time now.

[landonxjames]

Could this be a Result instead of a panic?

[aajtodd]

I left it as is for now. On one hand I agree on the other you defined the topology, punting for now, can revisit before main.

[landonxjames]

Error matching on the exact string value feels sketchy, will break unexpectedly is hyper ever updates this error. But maybe there is no way to get something better to match on?

[aajtodd]

Agreed, I opened hyperium/hyper#4119 for this, this is a workaround currently.

[aajtodd]

Upstream PR for vendored cache additions: hyperium/hyper-util#295