AWS SDK

/*

 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.

 * SPDX-License-Identifier: Apache-2.0

 */

//! Pool behavior tests parameterized over HTTP client implementations.

//!

//! Each test is written once as an `async fn` that takes a `&dyn MakeClient`,

//! then invoked for each client implementation.

#![cfg(all(feature = "wire-mock", feature = "default-client"))]

use aws_smithy_async::time::SystemTimeSource;

use aws_smithy_http_client::pool::{Client as PoolClient, SharedPool};

use aws_smithy_http_client::test_util::wire::connection::{

    ConnectionBehavior, ConnectionTestHarness,

};

use aws_smithy_http_client::test_util::wire::{ReplayedEvent, WireMockServer};

use aws_smithy_http_client::{ev, match_events, Builder, Connector};

use aws_smithy_runtime_api::client::http::{

    HttpClient, HttpConnector, HttpConnectorSettings, SharedHttpClient,

};

use aws_smithy_runtime_api::client::orchestrator::HttpRequest;

use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;

use aws_smithy_runtime_api::shared::IntoShared;

use std::borrow::Cow;

use std::net::{IpAddr, Ipv4Addr};

use std::time::Duration;

const IP1: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);

// ---------------------------------------------------------------------------

// ClientConfig + MakeClient

// ---------------------------------------------------------------------------

#[derive(Clone, Default)]

struct ClientConfig {

    idle_timeout: Option<Duration>,

    max_connections: Option<usize>,

    max_connections_per_host: Option<usize>,

}

impl ClientConfig {

    fn with_idle_timeout(mut self, timeout: Duration) -> Self {

        self.idle_timeout = Some(timeout);

        self

    }

    fn with_max_connections(mut self, n: usize) -> Self {

        self.max_connections = Some(n);

        self

    }

    fn with_max_connections_per_host(mut self, n: usize) -> Self {

        self.max_connections_per_host = Some(n);

        self

    }

}

trait MakeClient: Send + Sync {

    fn make(&self, config: ClientConfig) -> SharedHttpClient;

}

/// Current hyper 1.x client via `Builder::new()`

struct V1Client;

impl MakeClient for V1Client {

    fn make(&self, config: ClientConfig) -> SharedHttpClient {

        Builder::new().build_with_connector_fn(move |_settings, _components| {

            let mut builder = Connector::builder();

            if let Some(timeout) = config.idle_timeout {

                builder = builder.pool_idle_timeout(timeout);

            }

            // v1 does not support max_connections

            builder.build_http()

        })

    }

}

// V2 client backed by the composable connection pool

struct V2Client;

impl MakeClient for V2Client {

    fn make(&self, config: ClientConfig) -> SharedHttpClient {

        let mut builder = SharedPool::builder();

        if let Some(timeout) = config.idle_timeout {

            builder = builder.pool_idle_timeout(timeout);

        }

        if let Some(n) = config.max_connections {

            builder = builder.max_connections(n);

        }

        if let Some(n) = config.max_connections_per_host {

            builder = builder.max_connections_per_host(n);

        }

        let pool = builder.build_http();

        PoolClient::new(&pool).into_shared()

    }

}

// ---------------------------------------------------------------------------

// Helpers

// ---------------------------------------------------------------------------

fn runtime_components() -> aws_smithy_runtime_api::client::runtime_components::RuntimeComponents {

    RuntimeComponentsBuilder::for_tests()

        .with_time_source(Some(SystemTimeSource::new()))

        .build()

        .expect("valid runtime components")

}

async fn send_to(

    client: &SharedHttpClient,

    url: &str,

) -> Result<

    aws_smithy_runtime_api::client::orchestrator::HttpResponse,

    aws_smithy_runtime_api::client::result::ConnectorError,

> {

    let settings = HttpConnectorSettings::builder().build();

    let components = runtime_components();

    let connector = client.http_connector(&settings, &components);

    connector

        .call(HttpRequest::get(url).expect("valid HTTP request"))

        .await

}

/// Send a request and read the response body to completion.

///

/// This ensures the connection's body guard is released, returning the

/// connection to the pool. Returns `(status, body_bytes)`.

async fn send_and_read_body(

    client: &SharedHttpClient,

    url: &str,

) -> Result<(u16, Vec<u8>), aws_smithy_runtime_api::client::result::ConnectorError> {

    use http_body_util::BodyExt;

    let resp = send_to(client, url).await?;

    let status = resp.status().as_u16();

    let body = resp

        .into_body()

        .collect()

        .await

        .expect("body should be readable")

        .to_bytes()

        .to_vec();

    Ok((status, body))

}

fn localhost_url(server: &WireMockServer) -> String {

    let endpoint = server.endpoint_url();

    let port = endpoint

        .rsplit(':')

        .next()

        .expect("endpoint URL should contain port");

    format!("http://127.0.0.1:{port}/")

}

// ---------------------------------------------------------------------------

// Test implementations

// ---------------------------------------------------------------------------

/// Connection reuse via HTTP/1.1 keep-alive: sequential requests reuse one TCP connection.

async fn connection_reuse(make: &dyn MakeClient) {

    let server = WireMockServer::start(vec![

        ReplayedEvent::status(200),

        ReplayedEvent::status(200),

        ReplayedEvent::status(200),

    ])

    .await;

    let client = make.make(ClientConfig::default());

    let url = localhost_url(&server);

    for i in 1..=3 {

        let resp = send_to(&client, &url)

            .await

            .unwrap_or_else(|e| panic!("request {i} should succeed: {e}"));

        assert_eq!(resp.status().as_u16(), 200, "request {i} should return 200");

    }

    match_events!(ev!(connect), ev!(http(200)), ev!(http(200)), ev!(http(200)))(&server.events());

}

/// Idle timeout eviction: a connection idle past the timeout is discarded.

async fn idle_timeout_eviction(make: &dyn MakeClient) {

    let server =

        WireMockServer::start(vec![ReplayedEvent::status(200), ReplayedEvent::status(200)]).await;

    let idle_timeout = Duration::from_millis(100);

    let client = make.make(ClientConfig::default().with_idle_timeout(idle_timeout));

    let url = localhost_url(&server);

    let status = send_to(&client, &url)

        .await

        .expect("first request should succeed")

        .status()

        .as_u16();

    assert_eq!(status, 200);

    // Response (and its body guard) dropped here — connection returns to pool.

    tokio::time::sleep(idle_timeout * 2).await;

    let status = send_to(&client, &url)

        .await

        .expect("second request should succeed after idle eviction")

        .status()

        .as_u16();

    assert_eq!(status, 200);

    match_events!(ev!(connect), ev!(http(200)), ev!(connect), ev!(http(200)))(&server.events());

}

/// A server that resets on connect should surface as a connector error.

async fn connection_reset_returns_error(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(IP1, vec![ConnectionBehavior::ResetOnConnect])

        .build()

        .await;

    let client = make.make(ClientConfig::default());

    let url = format!("http://127.0.0.1:{}/", harness.endpoints[0].port());

    let err = send_to(&client, &url)

        .await

        .expect_err("request to a reset-on-connect endpoint should fail");

    assert!(err.is_io(), "expected ConnectorError::io, got: {err:?}");

}

/// The client should report correct connector metadata.

async fn connector_metadata(make: &dyn MakeClient) {

    let client = make.make(ClientConfig::default());

    let metadata = client

        .connector_metadata()

        .expect("connector_metadata should return Some");

    assert_eq!(metadata.name(), Cow::Borrowed("hyper"));

    assert_eq!(

        metadata.version(),

        Some(Cow::Borrowed("1.x")),

        "expected hyper 1.x connector version"

    );

}

// ---------------------------------------------------------------------------

// v1 test runners

// ---------------------------------------------------------------------------

#[tokio::test]

async fn v1_connection_reuse() {

    connection_reuse(&V1Client).await;

}

#[tokio::test]

async fn v1_idle_timeout_eviction() {

    idle_timeout_eviction(&V1Client).await;

}

#[tokio::test]

async fn v1_connection_reset_returns_error() {

    connection_reset_returns_error(&V1Client).await;

}

#[tokio::test]

async fn v1_connector_metadata() {

    connector_metadata(&V1Client).await;

}

// ---------------------------------------------------------------------------

// v2 test runners

// ---------------------------------------------------------------------------

#[tokio::test]

async fn v2_connection_reuse() {

    connection_reuse(&V2Client).await;

}

#[tokio::test]

async fn v2_idle_timeout_eviction() {

    // An idle connection is evicted after the pool idle timeout, so a request

    // arriving after the timeout opens a fresh connection. The body of the

    // first response is fully consumed, so its connection returns to the pool

    // and idles (the eviction target) rather than staying checked out.

    let idle_timeout = Duration::from_millis(100);

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"first",

                },

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"second",

                },

            ],

        )

        .build()

        .await;

    let pool = SharedPool::builder()

        .dns_resolver(harness.dns_resolver())

        .pool_idle_timeout(idle_timeout)

        .build_http();

    let client = PoolClient::new(&pool).into_shared();

    let url = format!("http://127.0.0.1:{}/", harness.endpoints[0].port());

    // First request: body consumed, connection returns to the pool idle. This

    // also lazily spawns the eviction task (pool_idle_timeout is set).

    let (status, _) = send_and_read_body(&client, &url)

        .await

        .expect("first request should succeed");

    assert_eq!(status, 200);

    assert_eq!(harness.tcp_accepted_count(), 1);

    // Wait past the idle timeout so the eviction task drops the idle connection.

    tokio::time::sleep(idle_timeout * 3).await;

    // The next request cannot reuse the evicted connection and opens a new one.

    let (status, _) = send_and_read_body(&client, &url)

        .await

        .expect("second request should succeed after idle eviction");

    assert_eq!(status, 200);

    assert_eq!(

        harness.tcp_accepted_count(),

        2,

        "the idle connection was evicted, so the second request reconnects"

    );

}

#[tokio::test]

async fn v2_connection_reset_returns_error() {

    connection_reset_returns_error(&V2Client).await;

}

#[tokio::test]

async fn v2_connector_metadata() {

    connector_metadata(&V2Client).await;

}

// ---------------------------------------------------------------------------

// Test implementations: origin-form URI + Host header

// ---------------------------------------------------------------------------

/// Requests should be sent with origin-form URI (just the path) and a correct Host header.

async fn origin_form_and_host_header(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![ConnectionBehavior::RespondKeepAlive {

                status: 200,

                body: b"ok",

            }],

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default());

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/some/path?key=val");

    let resp = send_to(&client, &url)

        .await

        .expect("request should succeed");

    assert_eq!(resp.status().as_u16(), 200);

    let requests = harness.http_requests();

    assert_eq!(requests.len(), 1, "expected exactly one HTTP request");

    let (uri, host) = &requests[0];

    // URI must be origin-form (path + query only, no scheme/authority)

    assert_eq!(uri, "/some/path?key=val", "URI should be origin-form");

    // Host header must be present with the correct authority

    let host = host.as_deref().expect("Host header should be present");

    assert_eq!(

        host,

        format!("127.0.0.1:{port}"),

        "Host header should match authority"

    );

}

#[tokio::test]

async fn v1_origin_form_and_host_header() {

    origin_form_and_host_header(&V1Client).await;

}

#[tokio::test]

async fn v2_origin_form_and_host_header() {

    origin_form_and_host_header(&V2Client).await;

}

// ---------------------------------------------------------------------------

// Test implementations: max_connections

// ---------------------------------------------------------------------------

/// With max_connections(2), concurrent requests should not open more than 2 connections,

/// and all requests should succeed (waiting requests served via connection reuse).

async fn max_connections_limits_concurrency(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            (0..10)

                .map(|_| ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"ok",

                })

                .collect(),

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default().with_max_connections(2));

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/");

    // Send 5 concurrent requests — all should succeed

    let mut tasks = tokio::task::JoinSet::new();

    for _ in 0..5 {

        let client = client.clone();

        let url = url.clone();

        tasks.spawn(async move { send_to(&client, &url).await });

    }

    let mut success_count = 0;

    while let Some(result) = tasks.join_next().await {

        let resp = result

            .expect("task should not panic")

            .expect("request should succeed");

        assert_eq!(resp.status().as_u16(), 200);

        success_count += 1;

    }

    assert_eq!(success_count, 5, "all 5 requests should complete");

    let accepted = harness.tcp_accepted_count();

    assert!(

        accepted <= 2,

        "expected at most 2 connections with max_connections(2), got {accepted}"

    );

}

/// Cached (reused) connections don't consume permits — sequential requests

/// with max_connections(1) all succeed on one connection.

async fn max_connections_reuse_does_not_consume_permits(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            (0..5)

                .map(|_| ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"ok",

                })

                .collect(),

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default().with_max_connections(1));

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/");

    // 5 sequential requests — all reuse the same connection

    for i in 0..5 {

        let resp = send_to(&client, &url)

            .await

            .unwrap_or_else(|e| panic!("request {i} should succeed: {e}"));

        assert_eq!(resp.status().as_u16(), 200);

    }

    assert_eq!(

        harness.tcp_accepted_count(),

        1,

        "all requests should reuse one connection"

    );

}

#[tokio::test]

async fn v2_max_connections_limits_concurrency() {

    max_connections_limits_concurrency(&V2Client).await;

}

#[tokio::test]

async fn v2_max_connections_reuse_does_not_consume_permits() {

    max_connections_reuse_does_not_consume_permits(&V2Client).await;

}

// ---------------------------------------------------------------------------

// Test implementations: max_connections_per_host

// ---------------------------------------------------------------------------

async fn is_bindable(ip: IpAddr) -> bool {

    tokio::net::TcpListener::bind((ip, 0u16)).await.is_ok()

}

const IP2: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2));

/// Per-host limit: each host gets its own budget.

async fn max_connections_per_host_limits_per_host(make: &dyn MakeClient) {

    if !is_bindable(IP2).await {

        eprintln!("skipping test: 127.0.0.2 not bindable");

        return;

    }

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            (0..10)

                .map(|_| ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"ok",

                })

                .collect(),

        )

        .endpoint(

            IP2,

            (0..10)

                .map(|_| ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"ok",

                })

                .collect(),

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default().with_max_connections_per_host(2));

    let port = harness.endpoints[0].port();

    // Send 4 concurrent requests to each host

    let mut tasks = tokio::task::JoinSet::new();

    for ip in ["127.0.0.1", "127.0.0.2"] {

        for _ in 0..4 {

            let client = client.clone();

            let url = format!("http://{ip}:{port}/");

            tasks.spawn(async move { send_to(&client, &url).await });

        }

    }

    while let Some(result) = tasks.join_next().await {

        result

            .expect("task should not panic")

            .expect("request should succeed");

    }

    let host1 = harness.tcp_accepted_by(IP1);

    let host2 = harness.tcp_accepted_by(IP2);

    assert!(host1 <= 2, "host 1: expected ≤2 connections, got {host1}");

    assert!(host2 <= 2, "host 2: expected ≤2 connections, got {host2}");

    // Total exceeds per-host limit (no global limit set)

    assert!(host1 + host2 > 2, "total should exceed per-host limit");

}

/// Global and per-host limits compose: global(3) + per_host(2) with 2 hosts.

async fn max_connections_global_and_per_host_compose(make: &dyn MakeClient) {

    if !is_bindable(IP2).await {

        eprintln!("skipping test: 127.0.0.2 not bindable");

        return;

    }

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            (0..10)

                .map(|_| ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"ok",

                })

                .collect(),

        )

        .endpoint(

            IP2,

            (0..10)

                .map(|_| ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"ok",

                })

                .collect(),

        )

        .build()

        .await;

    let client = make.make(

        ClientConfig::default()

            .with_max_connections(3)

            .with_max_connections_per_host(2),

    );

    let port = harness.endpoints[0].port();

    // Send 4 concurrent requests to each host (8 total)

    let mut tasks = tokio::task::JoinSet::new();

    for ip in ["127.0.0.1", "127.0.0.2"] {

        for _ in 0..4 {

            let client = client.clone();

            let url = format!("http://{ip}:{port}/");

            tasks.spawn(async move { send_to(&client, &url).await });

        }

    }

    while let Some(result) = tasks.join_next().await {

        result

            .expect("task should not panic")

            .expect("request should succeed");

    }

    let host1 = harness.tcp_accepted_by(IP1);

    let host2 = harness.tcp_accepted_by(IP2);

    // Per-host: each ≤2

    assert!(host1 <= 2, "host 1: expected ≤2, got {host1}");

    assert!(host2 <= 2, "host 2: expected ≤2, got {host2}");

    // Global: total ≤3

    assert!(

        host1 + host2 <= 3,

        "total: expected ≤3 (global limit), got {}",

        host1 + host2

    );

}

#[tokio::test]

async fn v2_max_connections_per_host_limits_per_host() {

    max_connections_per_host_limits_per_host(&V2Client).await;

}

#[tokio::test]

async fn v2_max_connections_global_and_per_host_compose() {

    max_connections_global_and_per_host_compose(&V2Client).await;

}

// ---------------------------------------------------------------------------

// Test implementations: connection lifecycle

// ---------------------------------------------------------------------------

/// Without max_connections, connections scale freely under concurrency.

async fn no_limit_allows_unbounded_connections(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            (0..20)

                .map(|_| ConnectionBehavior::HoldThenClose(Duration::from_secs(2)))

                .collect(),

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default());

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/");

    // Send 5 concurrent requests — server holds each connection open,

    // so the client must open 5 separate connections.

    let mut tasks = tokio::task::JoinSet::new();

    for _ in 0..5 {

        let client = client.clone();

        let url = url.clone();

        tasks.spawn(async move {

            // These will fail (server doesn't send HTTP response), but

            // the point is that 5 TCP connections are opened concurrently.

            let _ = send_to(&client, &url).await;

        });

    }

    // Give time for all connections to be established

    tokio::time::sleep(Duration::from_millis(200)).await;

    let accepted = harness.tcp_accepted_count();

    assert!(

        accepted >= 4,

        "without max_connections, expected ≥4 concurrent connections, got {accepted}"

    );

    // Clean up tasks

    tasks.shutdown().await;

}

/// Sequential requests reuse the same connection (body fully consumed between requests).

async fn connection_reuse_after_body_consumed(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            (0..3)

                .map(|_| ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"hello world response body",

                })

                .collect(),

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default());

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/");

    for i in 0..3 {

        let (status, body) = send_and_read_body(&client, &url)

            .await

            .unwrap_or_else(|e| panic!("request {i} should succeed: {e}"));

        assert_eq!(status, 200);

        assert_eq!(body, b"hello world response body");

    }

    assert_eq!(

        harness.tcp_accepted_count(),

        1,

        "all requests should reuse one connection after body is consumed"

    );

}

/// Back-to-back requests reuse one connection despite a short idle timeout:

/// each re-checkout beats the eviction tick, so no reconnect occurs.

async fn active_connection_survives_idle_timeout(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"first",

                },

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"second",

                },

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"third",

                },

            ],

        )

        .build()

        .await;

    // Use a very short idle timeout — but sequential requests should still

    // reuse the connection because the pool doesn't evict between requests

    // that happen back-to-back.

    let client = make.make(ClientConfig::default().with_idle_timeout(Duration::from_millis(100)));

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/");

    // Three back-to-back requests — all should reuse the same connection

    let expected = [b"first".as_slice(), b"second", b"third"];

    for (i, expected_body) in expected.iter().enumerate() {

        let (status, body) = send_and_read_body(&client, &url)

            .await

            .unwrap_or_else(|e| panic!("request {i} should succeed: {e}"));

        assert_eq!(status, 200);

        assert_eq!(body, *expected_body, "request {i} body mismatch");

    }

    assert_eq!(

        harness.tcp_accepted_count(),

        1,

        "back-to-back requests should reuse one connection even with short idle timeout"

    );

}

#[tokio::test]

async fn v2_no_limit_allows_unbounded_connections() {

    no_limit_allows_unbounded_connections(&V2Client).await;

}

#[tokio::test]

async fn v2_connection_reuse_after_body_consumed() {

    connection_reuse_after_body_consumed(&V2Client).await;

}

#[tokio::test]

async fn v2_active_connection_survives_idle_timeout() {

    active_connection_survives_idle_timeout(&V2Client).await;

}

// ---------------------------------------------------------------------------

// Test implementations: connection lifecycle

// ---------------------------------------------------------------------------

/// Server closes an idle connection (simulating server-side idle timeout).

/// The connection returns to the pool clean, then dies while idle. On next

/// checkout, poll_ready detects the dead connection, the checkout loop

/// discards it, and a fresh connection is created.

async fn stale_connection_detected_at_checkout(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![

                // First connection: respond with keep-alive, then close after 30ms

                ConnectionBehavior::RespondThenIdleClose {

                    status: 200,

                    body: b"first",

                    idle: Duration::from_millis(30),

                },

                // Second connection (after stale one is discarded)

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"second",

                },

            ],

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default());

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/");

    // First request succeeds, body consumed, connection returns to pool

    let (status, body) = send_and_read_body(&client, &url)

        .await

        .expect("first request");

    assert_eq!(status, 200);

    assert_eq!(body, b"first");

    // Wait for server to close the idle connection (30ms idle + margin)

    tokio::time::sleep(Duration::from_millis(80)).await;

    // Second request: checkout finds stale connection, discards, creates new

    let (status, body) = send_and_read_body(&client, &url)

        .await

        .expect("second request should succeed on fresh connection");

    assert_eq!(status, 200);

    assert_eq!(body, b"second");

    assert_eq!(

        harness.tcp_accepted_count(),

        2,

        "should have opened 2 connections (first died while idle in pool)"

    );

}

/// When the response body is not consumed, the connection is held by the

/// body guard and unavailable for reuse. A concurrent request must open

/// a new connection.

async fn unconsumed_body_holds_connection(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"first",

                },

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"second",

                },

            ],

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default());

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/");

    // First request — hold the response (body unconsumed, connection held)

    let _held_resp = send_to(&client, &url).await.expect("first request");

    // Second request while first response is still held — must open new connection

    let (status, body) = send_and_read_body(&client, &url)

        .await

        .expect("second request");

    assert_eq!(status, 200);

    assert_eq!(body, b"second");

    assert_eq!(

        harness.tcp_accepted_count(),

        2,

        "second request should open a new connection while first body is held"

    );

    // Drop the held response — connection guard released

    drop(_held_resp);

}

/// Server sends Connection: close — client should not reuse the connection.

async fn connection_close_header_prevents_reuse(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![

                ConnectionBehavior::RespondThenClose {

                    status: 200,

                    body: b"closing",

                },

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"fresh",

                },

            ],

        )

        .build()

        .await;

    let client = make.make(ClientConfig::default());

    let port = harness.endpoints[0].port();

    let url = format!("http://127.0.0.1:{port}/");

    let (status, body) = send_and_read_body(&client, &url)

        .await

        .expect("first request");

    assert_eq!(status, 200);

    assert_eq!(body, b"closing");

    // Connection: close means the client should not attempt to reuse

    let (status, body) = send_and_read_body(&client, &url)

        .await

        .expect("second request");

    assert_eq!(status, 200);

    assert_eq!(body, b"fresh");

    assert_eq!(

        harness.tcp_accepted_count(),

        2,

        "Connection: close should prevent reuse"

    );

}

#[tokio::test]

async fn v2_stale_connection_detected_at_checkout() {

    stale_connection_detected_at_checkout(&V2Client).await;

}

#[tokio::test]

async fn v2_unconsumed_body_holds_connection() {

    unconsumed_body_holds_connection(&V2Client).await;

}

#[tokio::test]

async fn v2_connection_close_header_prevents_reuse() {

    connection_close_header_prevents_reuse(&V2Client).await;

}

// ---------------------------------------------------------------------------

// Test implementations: connection poisoning

// ---------------------------------------------------------------------------

/// Send a request and return both the response body and the `ConnectionMetadata`

/// captured by a `CaptureSmithyConnection` attached to the request. This is the

/// integration surface the `ConnectionPoisoningInterceptor` uses in real SDK flows

/// — the adapter must populate a retriever that returns live metadata pointing

/// at the connection selected for this request.

async fn send_with_capture(

    client: &SharedHttpClient,

    url: &str,

) -> (

    u16,

    Vec<u8>,

    Option<aws_smithy_runtime_api::client::connection::ConnectionMetadata>,

) {

    use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;

    use http_body_util::BodyExt;

    let settings = HttpConnectorSettings::builder().build();

    let components = runtime_components();

    let connector = client.http_connector(&settings, &components);

    let capture = CaptureSmithyConnection::new();

    let mut request = HttpRequest::get(url).expect("valid HTTP request");

    request.add_extension(capture.clone());

    let resp = connector

        .call(request)

        .await

        .expect("request should succeed");

    let status = resp.status().as_u16();

    let body = resp

        .into_body()

        .collect()

        .await

        .expect("body should be readable")

        .to_bytes()

        .to_vec();

    (status, body, capture.get())

}

/// Poisoning a connection prevents it from being reused.

///

/// Mirrors the production flow: `ConnectionPoisoningInterceptor` attaches a

/// `CaptureSmithyConnection` to the request, the adapter populates it with

/// metadata pointing at the selected connection, and on a transient error

/// the interceptor calls `ConnectionMetadata::poison()`. The next request

/// for the same host must establish a new TCP connection instead of reusing

/// the poisoned one.

async fn poisoned_connection_not_reused(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"first",

                },

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"second",

                },

            ],

        )

        .build()

        .await;

    let url = format!("http://127.0.0.1:{}/", harness.endpoints[0].port());

    let client = make.make(ClientConfig::default());

    // First request: establish connection, capture metadata.

    let (status, body, metadata) = send_with_capture(&client, &url).await;

    assert_eq!(status, 200);

    assert_eq!(body, b"first");

    let metadata = metadata.expect("adapter should populate CaptureSmithyConnection");

    assert_eq!(

        harness.tcp_accepted_count(),

        1,

        "first request opens one connection"

    );

    // Poison the connection — what the orchestrator does on a transient error.

    metadata.poison();

    // Next request to the same host must open a NEW connection; the poisoned

    // one is skipped on checkout and dropped on return.

    let (status, body, _) = send_with_capture(&client, &url).await;

    assert_eq!(status, 200);

    assert_eq!(body, b"second");

    assert_eq!(

        harness.tcp_accepted_count(),

        2,

        "poisoned connection must not be reused"

    );

}

/// When no transient error occurs, the captured metadata is handed out but

/// never poisoned — the connection returns to the pool and is reused.

/// This is the non-poison control for `poisoned_connection_not_reused`.

async fn capture_without_poison_allows_reuse(make: &dyn MakeClient) {

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"first",

                },

                ConnectionBehavior::RespondKeepAlive {

                    status: 200,

                    body: b"second",

                },

            ],

        )

        .build()

        .await;

    let url = format!("http://127.0.0.1:{}/", harness.endpoints[0].port());

    let client = make.make(ClientConfig::default());

    let (status, body, metadata) = send_with_capture(&client, &url).await;

    assert_eq!(status, 200);

    assert_eq!(body, b"first");

    assert!(

        metadata.is_some(),

        "adapter should populate CaptureSmithyConnection"

    );

    // Deliberately not calling poison().

    drop(metadata);

    let (status, body, _) = send_with_capture(&client, &url).await;

    assert_eq!(status, 200);

    assert_eq!(body, b"second");

    assert_eq!(

        harness.tcp_accepted_count(),

        1,

        "without poison the connection should be reused"

    );

}

#[tokio::test]

async fn v2_poisoned_connection_not_reused() {

    poisoned_connection_not_reused(&V2Client).await;

}

#[tokio::test]

async fn v2_capture_without_poison_allows_reuse() {

    capture_without_poison_allows_reuse(&V2Client).await;

}

// ---------------------------------------------------------------------------

// Test implementations: per-operation timeouts

// ---------------------------------------------------------------------------

/// Read timeout fires when the server accepts TCP but never sends a response.

///

/// Exercises the v2 adapter's per-op timeout wrapping end-to-end: the

/// `HttpConnectorSettings::read_timeout` flows through

/// `HttpClient::http_connector` into `PooledConnector`, wraps

/// `pool.send_request`, fires because `HoldThenClose` never replies, and

/// produces a `ConnectorError::timeout` classified by `downcast_error`.

async fn v2_read_timeout_fires_on_silent_server() {

    use aws_smithy_async::rt::sleep::{SharedAsyncSleep, TokioSleep};

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![ConnectionBehavior::HoldThenClose(Duration::from_secs(30))],

        )

        .build()

        .await;

    let client = V2Client.make(ClientConfig::default());

    let components =

        aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder::for_tests()

            .with_time_source(Some(SystemTimeSource::new()))

            .with_sleep_impl(Some(SharedAsyncSleep::new(TokioSleep::new())))

            .build()

            .expect("valid runtime components");

    let settings = HttpConnectorSettings::builder()

        .read_timeout(Duration::from_millis(200))

        .build();

    let connector = client.http_connector(&settings, &components);

    let url = format!("http://127.0.0.1:{}/", harness.endpoints[0].port());

    let start = std::time::Instant::now();

    let err = connector

        .call(HttpRequest::get(&url).expect("valid HTTP request"))

        .await

        .expect_err("read timeout should fire against a non-responsive server");

    let elapsed = start.elapsed();

    assert!(

        err.is_timeout(),

        "expected timeout classification, got {err:?}"

    );

    assert!(

        elapsed < Duration::from_secs(2),

        "read timeout did not fire in time (took {elapsed:?})"

    );

}

#[tokio::test]

async fn v2_read_timeout() {

    v2_read_timeout_fires_on_silent_server().await;

}

// ---------------------------------------------------------------------------

// Tracing output assertions

// ---------------------------------------------------------------------------

//

// These tests verify the pool emits useful structured tracing events.

// They use a thread-local subscriber and `current_thread` tokio so spawned

// tasks (eviction) run on the same thread. Must be run with

// `--test-threads=1` to avoid subscriber interference from parallel tests.

fn capture_pool_logs() -> (

    tracing::subscriber::DefaultGuard,

    std::sync::Arc<std::sync::Mutex<Vec<u8>>>,

) {

    use std::io::Write;

    use std::sync::{Arc, Mutex};

    struct BufWriter(Arc<Mutex<Vec<u8>>>);

    impl Write for BufWriter {

        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {

            self.0.lock().unwrap().extend_from_slice(buf);

            Ok(buf.len())

        }

        fn flush(&mut self) -> std::io::Result<()> {

            Ok(())

        }

    }

    let buf = Arc::new(Mutex::new(Vec::<u8>::new()));

    let buf_clone = buf.clone();

    let subscriber = tracing_subscriber::fmt()

        .with_ansi(false)

        .with_max_level(tracing::Level::TRACE)

        .with_writer(move || BufWriter(buf_clone.clone()))

        .finish();

    let guard = tracing::subscriber::set_default(subscriber);

    (guard, buf)

}

fn captured_str(buf: &std::sync::Arc<std::sync::Mutex<Vec<u8>>>) -> String {

    String::from_utf8(buf.lock().unwrap().clone()).expect("captured logs are utf-8")

}

/// Background eviction task emits structured tracing events including

/// pool init, connection established, eviction with reason, and host

/// entry removal.

///

/// Requires `--test-threads=1` due to thread-local subscriber; run via:

/// `cargo test --features default-client,test-util,wire-mock --test pool_behavior_test -- --ignored --test-threads=1`

#[tokio::test(flavor = "current_thread")]

#[serial_test::serial(tracing)]

#[ignore]

async fn v2_background_eviction_emits_tracing_events() {

    let (_guard, logs) = capture_pool_logs();

    let server = WireMockServer::start(vec![ReplayedEvent::status(200)]).await;

    let idle_timeout = Duration::from_millis(100);

    let client = V2Client.make(ClientConfig::default().with_idle_timeout(idle_timeout));

    let url = localhost_url(&server);

    let status = send_to(&client, &url)

        .await

        .expect("first request should succeed")

        .status()

        .as_u16();

    assert_eq!(status, 200);

    // Wait past idle timeout + eviction tick intervals.

    tokio::time::sleep(idle_timeout * 5).await;

    let captured = captured_str(&logs);

    assert!(

        captured.contains("pool: initialized"),

        "expected pool init log. captured:\n{captured}"

    );

    assert!(

        captured.contains("pool: eviction task spawned"),

        "expected eviction task spawn log. captured:\n{captured}"

    );

    assert!(

        captured.contains("pool: connection established"),

        "expected connection-established log. captured:\n{captured}"

    );

    assert!(

        captured.contains("pool: connection evicted"),

        "expected connection-evicted log. captured:\n{captured}"

    );

    assert!(

        captured.contains("pool: host entry removed"),

        "expected host-entry-removed log. captured:\n{captured}"

    );

}

/// conn_id is stable (same across a connection's lifetime) and monotonically

/// increasing (new connection after eviction gets the next id).

///

/// Requires `--test-threads=1`; see `v2_background_eviction_emits_tracing_events`.

#[tokio::test(flavor = "current_thread")]

#[serial_test::serial(tracing)]

#[ignore]

async fn v2_conn_id_is_stable_and_monotonic() {

    let (_guard, logs) = capture_pool_logs();

    let server =

        WireMockServer::start(vec![ReplayedEvent::status(200), ReplayedEvent::status(200)]).await;

    let idle_timeout = Duration::from_millis(100);

    let client = V2Client.make(ClientConfig::default().with_idle_timeout(idle_timeout));

    let url = localhost_url(&server);

    send_to(&client, &url).await.expect("first request");

    tokio::time::sleep(idle_timeout * 5).await;

    send_to(&client, &url).await.expect("second request");

    let captured = captured_str(&logs);

    assert!(

        captured.contains("conn_id=0"),

        "first connection should be conn_id=0. captured:\n{captured}"

    );

    assert!(

        captured.contains("conn_id=1"),

        "second connection (after eviction) should be conn_id=1. captured:\n{captured}"

    );

}

// ---------------------------------------------------------------------------

// Connect timeout

// ---------------------------------------------------------------------------

/// Proves that `connect_timeout` from `HttpConnectorSettings` fires when the

/// TCP connection cannot be established within the deadline. Uses TEST-NET-1

/// (192.0.2.1), a non-routable address guaranteed to black-hole SYN packets.

#[tokio::test]

async fn v2_connect_timeout() {

    use aws_smithy_async::rt::sleep::{SharedAsyncSleep, TokioSleep};

    let client = V2Client.make(ClientConfig::default());

    let components =

        aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder::for_tests()

            .with_time_source(Some(SystemTimeSource::new()))

            .with_sleep_impl(Some(SharedAsyncSleep::new(TokioSleep::new())))

            .build()

            .expect("valid runtime components");

    let settings = HttpConnectorSettings::builder()

        .connect_timeout(Duration::from_millis(500))

        .build();

    let connector = client.http_connector(&settings, &components);

    let start = std::time::Instant::now();

    let err = connector

        .call(HttpRequest::get("http://192.0.2.1:1234/unreachable").expect("valid request"))

        .await

        .expect_err("connect timeout should fire against non-routable address");

    let elapsed = start.elapsed();

    assert!(

        err.is_timeout(),

        "expected timeout classification, got {err:?}"

    );

    assert!(

        elapsed >= Duration::from_millis(450),

        "timeout fired too early ({elapsed:?})"

    );

    assert!(

        elapsed < Duration::from_secs(5),

        "timeout took too long ({elapsed:?})"

    );

}

/// `ConnectionMetadata::connection_id()` surfaces the pool-assigned id

/// through `CaptureSmithyConnection`. Sequential requests on the same

/// connection share the same id; a new connection after eviction gets a

/// different id.

#[tokio::test]

async fn v2_connection_id_surfaced_through_metadata() {

    use aws_smithy_runtime_api::client::connection::ConnectionId;

    let harness = ConnectionTestHarness::builder()

        .endpoint(

            IP1,

            vec![