asupersync 0.3.0

//! HTTP connection pooling.
//!
//! Provides connection pool management for HTTP clients, enabling connection
//! reuse and efficient handling of concurrent requests to the same hosts.
//!
//! # Connection Reuse
//!
//! The pool maintains idle connections that can be reused for subsequent
//! requests to the same host, reducing connection establishment overhead.
//!
//! # Pool Configuration
//!
//! The pool can be configured with:
//! - Maximum connections per host
//! - Maximum total connections
//! - Idle connection timeout
//! - Connection health checks

use smallvec::SmallVec;
use std::collections::HashMap;
use std::time::Duration;

use crate::types::Time;

/// Connection pool key identifying a specific host.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct PoolKey {
    /// Host name or IP address.
    pub host: String,
    /// Port number.
    pub port: u16,
    /// Whether this is an HTTPS connection.
    pub is_https: bool,
}

impl PoolKey {
    /// Creates a new pool key.
    #[must_use]
    pub fn new(host: impl Into<String>, port: u16, is_https: bool) -> Self {
        Self {
            host: host.into(),
            port,
            is_https,
        }
    }

    /// Creates a pool key for HTTP (port 80 default).
    #[must_use]
    pub fn http(host: impl Into<String>, port: Option<u16>) -> Self {
        Self::new(host, port.unwrap_or(80), false)
    }

    /// Creates a pool key for HTTPS (port 443 default).
    #[must_use]
    pub fn https(host: impl Into<String>, port: Option<u16>) -> Self {
        Self::new(host, port.unwrap_or(443), true)
    }
}

/// Configuration for the connection pool.
#[derive(Debug, Clone)]
pub struct PoolConfig {
    /// Maximum connections per host.
    pub max_connections_per_host: usize,
    /// Maximum total connections across all hosts.
    pub max_total_connections: usize,
    /// How long an idle connection is kept before eviction.
    pub idle_timeout: Duration,
    /// How often to run the idle connection cleanup.
    pub cleanup_interval: Duration,
}

impl Default for PoolConfig {
    fn default() -> Self {
        Self {
            max_connections_per_host: 6,
            max_total_connections: 100,
            idle_timeout: Duration::from_secs(90),
            cleanup_interval: Duration::from_secs(30),
        }
    }
}

impl PoolConfig {
    /// Creates a new pool configuration builder.
    #[must_use]
    pub fn builder() -> PoolConfigBuilder {
        PoolConfigBuilder::default()
    }
}

/// Builder for [`PoolConfig`].
#[derive(Debug, Default)]
pub struct PoolConfigBuilder {
    max_connections_per_host: Option<usize>,
    max_total_connections: Option<usize>,
    idle_timeout: Option<Duration>,
    cleanup_interval: Option<Duration>,
}

impl PoolConfigBuilder {
    /// Sets the maximum connections per host.
    #[must_use]
    pub fn max_connections_per_host(mut self, max: usize) -> Self {
        self.max_connections_per_host = Some(max);
        self
    }

    /// Sets the maximum total connections.
    #[must_use]
    pub fn max_total_connections(mut self, max: usize) -> Self {
        self.max_total_connections = Some(max);
        self
    }

    /// Sets the idle connection timeout.
    #[must_use]
    pub fn idle_timeout(mut self, timeout: Duration) -> Self {
        self.idle_timeout = Some(timeout);
        self
    }

    /// Sets the cleanup interval.
    #[must_use]
    pub fn cleanup_interval(mut self, interval: Duration) -> Self {
        self.cleanup_interval = Some(interval);
        self
    }

    /// Builds the configuration.
    #[must_use]
    pub fn build(self) -> PoolConfig {
        let defaults = PoolConfig::default();
        PoolConfig {
            max_connections_per_host: self
                .max_connections_per_host
                .unwrap_or(defaults.max_connections_per_host),
            max_total_connections: self
                .max_total_connections
                .unwrap_or(defaults.max_total_connections),
            idle_timeout: self.idle_timeout.unwrap_or(defaults.idle_timeout),
            cleanup_interval: self.cleanup_interval.unwrap_or(defaults.cleanup_interval),
        }
    }
}

/// Connection state in the pool.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PooledConnectionState {
    /// Connection is idle and available for reuse.
    Idle,
    /// Connection is currently in use.
    InUse,
    /// Connection is being established.
    Connecting,
    /// Connection has been marked as unhealthy.
    Unhealthy,
}

/// Metadata for a pooled connection.
#[derive(Debug)]
pub struct PooledConnectionMeta {
    /// Unique identifier for this connection.
    pub id: u64,
    /// Connection state.
    pub state: PooledConnectionState,
    /// When the connection was created.
    pub created_at: Time,
    /// When the connection was last used.
    pub last_used: Time,
    /// Number of requests served by this connection.
    pub requests_served: u64,
    /// HTTP version (1 for HTTP/1.x, 2 for HTTP/2).
    pub http_version: u8,
}

impl PooledConnectionMeta {
    /// Creates new connection metadata.
    #[must_use]
    pub fn new(id: u64, now: Time, http_version: u8) -> Self {
        Self {
            id,
            state: PooledConnectionState::Connecting,
            created_at: now,
            last_used: now,
            requests_served: 0,
            http_version,
        }
    }

    /// Marks the connection as idle.
    pub fn mark_idle(&mut self, now: Time) {
        self.state = PooledConnectionState::Idle;
        self.last_used = now;
    }

    /// Marks the connection as connected only if it is still connecting.
    ///
    /// Returns true when the transition was applied.
    #[must_use]
    pub fn mark_connected(&mut self, now: Time) -> bool {
        if self.state == PooledConnectionState::Connecting {
            self.mark_idle(now);
            true
        } else {
            false
        }
    }

    /// Marks the connection as in use.
    pub fn mark_in_use(&mut self) {
        self.state = PooledConnectionState::InUse;
        self.requests_served += 1;
    }

    /// Returns the connection to idle only if it is currently checked out.
    ///
    /// Returns true when the transition was applied.
    #[must_use]
    pub fn release(&mut self, now: Time) -> bool {
        if self.state == PooledConnectionState::InUse {
            self.mark_idle(now);
            true
        } else {
            false
        }
    }

    /// Marks the connection as unhealthy.
    pub fn mark_unhealthy(&mut self) {
        self.state = PooledConnectionState::Unhealthy;
    }

    /// Returns true if this connection has been idle longer than the timeout.
    #[must_use]
    pub fn is_expired(&self, now: Time, idle_timeout: Duration) -> bool {
        if self.state != PooledConnectionState::Idle {
            return false;
        }
        let elapsed_nanos = now.duration_since(self.last_used);
        elapsed_nanos >= u64::try_from(idle_timeout.as_nanos()).unwrap_or(u64::MAX)
    }
}

/// Statistics for a connection pool.
#[derive(Debug, Default, Clone)]
pub struct PoolStats {
    /// Total number of connections currently in the pool.
    pub total_connections: usize,
    /// Number of idle connections.
    pub idle_connections: usize,
    /// Number of connections in use.
    pub in_use_connections: usize,
    /// Number of connections being established.
    pub connecting: usize,
    /// Number of hosts with connections.
    pub hosts_count: usize,
    /// Total connections created over the pool's lifetime.
    pub connections_created: u64,
    /// Total connections closed over the pool's lifetime.
    pub connections_closed: u64,
    /// Total connections that timed out.
    pub connections_timed_out: u64,
}

/// Tracks connections for a single host.
#[derive(Debug)]
struct HostPool {
    /// Connections for this host (by connection ID).
    connections: HashMap<u64, PooledConnectionMeta>,
}

impl HostPool {
    fn new() -> Self {
        Self {
            connections: HashMap::new(),
        }
    }

    fn connection_count(&self) -> usize {
        self.connections.len()
    }

    fn idle_count(&self) -> usize {
        self.connections
            .values()
            .filter(|c| c.state == PooledConnectionState::Idle)
            .count()
    }

    fn in_use_count(&self) -> usize {
        self.connections
            .values()
            .filter(|c| c.state == PooledConnectionState::InUse)
            .count()
    }

    fn connecting_count(&self) -> usize {
        self.connections
            .values()
            .filter(|c| c.state == PooledConnectionState::Connecting)
            .count()
    }
}

/// HTTP connection pool.
///
/// Manages a pool of connections to different hosts, enabling connection
/// reuse for improved performance.
#[derive(Debug)]
pub struct Pool {
    /// Pool configuration.
    config: PoolConfig,
    /// Connections organized by host.
    hosts: HashMap<PoolKey, HostPool>,
    /// Next connection ID.
    next_id: u64,
    /// Lifetime statistics.
    stats: PoolStats,
    /// Last time cleanup was run.
    last_cleanup: Time,
}

impl Pool {
    /// Creates a new connection pool with default configuration.
    #[must_use]
    pub fn new() -> Self {
        Self::with_config(PoolConfig::default())
    }

    /// Creates a new connection pool with the given configuration.
    #[must_use]
    pub fn with_config(config: PoolConfig) -> Self {
        Self {
            config,
            hosts: HashMap::new(),
            next_id: 1,
            stats: PoolStats::default(),
            last_cleanup: Time::ZERO,
        }
    }

    /// Returns the pool configuration.
    #[must_use]
    pub fn config(&self) -> &PoolConfig {
        &self.config
    }

    /// Returns current pool statistics.
    #[must_use]
    pub fn stats(&self) -> PoolStats {
        let mut stats = self.stats.clone();
        stats.hosts_count = self.hosts.len();
        stats.total_connections = 0;
        stats.idle_connections = 0;
        stats.in_use_connections = 0;
        stats.connecting = 0;

        for host_pool in self.hosts.values() {
            stats.total_connections += host_pool.connection_count();
            stats.idle_connections += host_pool.idle_count();
            stats.in_use_connections += host_pool.in_use_count();
            stats.connecting += host_pool.connecting_count();
        }

        stats
    }

    /// Attempts to acquire an idle connection for the given key.
    ///
    /// Returns the connection ID if an idle connection is available.
    pub fn try_acquire(&mut self, key: &PoolKey, now: Time) -> Option<u64> {
        self.maybe_cleanup(now);
        let host_pool = self.hosts.get_mut(key)?;

        // HashMap iteration order is randomized, so explicitly choose the
        // lowest viable connection ID to keep reuse deterministic.
        let idle_id = host_pool
            .connections
            .iter()
            .filter(|(_, conn)| {
                conn.state == PooledConnectionState::Idle
                    && !conn.is_expired(now, self.config.idle_timeout)
            })
            .map(|(id, _)| *id)
            .min();

        if let Some(id) = idle_id {
            if let Some(conn) = host_pool.connections.get_mut(&id) {
                conn.mark_in_use();
                return Some(id);
            }
        }

        None
    }

    /// Checks if a new connection can be created for the given key.
    ///
    /// This also evicts expired idle connections using the provided timestamp
    /// so stale entries do not block new connections.
    #[must_use]
    pub fn can_create_connection(&mut self, key: &PoolKey, now: Time) -> bool {
        // Capacity checks must ignore expired idle sockets immediately, not only
        // on periodic cleanup ticks.
        self.cleanup_expired(now);
        self.last_cleanup = now;
        // Check total connection limit
        let total = self
            .hosts
            .values()
            .map(HostPool::connection_count)
            .sum::<usize>();
        if total >= self.config.max_total_connections {
            return false;
        }

        // Check per-host limit
        if let Some(host_pool) = self.hosts.get(key) {
            if host_pool.connection_count() >= self.config.max_connections_per_host {
                return false;
            }
        }

        true
    }

    fn maybe_cleanup(&mut self, now: Time) {
        let elapsed = now.as_nanos().saturating_sub(self.last_cleanup.as_nanos());
        let interval_nanos =
            u64::try_from(self.config.cleanup_interval.as_nanos()).unwrap_or(u64::MAX);
        if elapsed >= interval_nanos {
            self.cleanup_expired(now);
            self.last_cleanup = now;
        }
    }

    /// Registers a new connection being established.
    ///
    /// Returns the connection ID for tracking.
    pub fn register_connecting(&mut self, key: PoolKey, now: Time, http_version: u8) -> u64 {
        let id = self.next_id;
        self.next_id = self
            .next_id
            .checked_add(1)
            .expect("http connection pool id counter exhausted");

        let meta = PooledConnectionMeta::new(id, now, http_version);

        let host_pool = self.hosts.entry(key).or_insert_with(HostPool::new);
        host_pool.connections.insert(id, meta);

        self.stats.connections_created += 1;

        id
    }

    /// Marks a connection as successfully established and idle.
    ///
    /// Returns true when the transition was applied.
    pub fn mark_connected(&mut self, key: &PoolKey, id: u64, now: Time) -> bool {
        if let Some(host_pool) = self.hosts.get_mut(key) {
            if let Some(conn) = host_pool.connections.get_mut(&id) {
                return conn.mark_connected(now);
            }
        }
        false
    }

    /// Returns a connection to the pool (makes it idle).
    ///
    /// Returns true when the transition was applied.
    pub fn release(&mut self, key: &PoolKey, id: u64, now: Time) -> bool {
        if let Some(host_pool) = self.hosts.get_mut(key) {
            if let Some(conn) = host_pool.connections.get_mut(&id) {
                return conn.release(now);
            }
        }
        false
    }

    /// Removes a connection from the pool.
    pub fn remove(&mut self, key: &PoolKey, id: u64) {
        if let Some(host_pool) = self.hosts.get_mut(key) {
            if host_pool.connections.remove(&id).is_some() {
                self.stats.connections_closed += 1;
            }

            // Clean up empty host pools
            if host_pool.connections.is_empty() {
                self.hosts.remove(key);
            }
        }
    }

    /// Removes expired idle connections and returns the retired `(key, id)` pairs.
    pub fn cleanup_expired_entries(&mut self, now: Time) -> Vec<(PoolKey, u64)> {
        let idle_timeout = self.config.idle_timeout;
        let mut removed = Vec::new();
        let mut empty_keys: SmallVec<[PoolKey; 4]> = SmallVec::new();

        for (key, host_pool) in &mut self.hosts {
            let expired_ids: SmallVec<[u64; 8]> = host_pool
                .connections
                .iter()
                .filter(|(_, conn)| conn.is_expired(now, idle_timeout))
                .map(|(id, _)| *id)
                .collect();

            for id in expired_ids {
                host_pool.connections.remove(&id);
                self.stats.connections_closed += 1;
                self.stats.connections_timed_out += 1;
                removed.push((key.clone(), id));
            }

            if host_pool.connections.is_empty() {
                empty_keys.push(key.clone());
            }
        }

        for key in empty_keys {
            self.hosts.remove(&key);
        }

        removed
    }

    /// Cleans up expired idle connections.
    ///
    /// Returns the number of connections removed.
    pub fn cleanup_expired(&mut self, now: Time) -> usize {
        self.cleanup_expired_entries(now).len()
    }

    /// Gets metadata for a specific connection.
    #[must_use]
    pub fn get_connection_meta(&self, key: &PoolKey, id: u64) -> Option<&PooledConnectionMeta> {
        self.hosts.get(key)?.connections.get(&id)
    }
}

impl Default for Pool {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn make_time(ms: u64) -> Time {
        Time::from_millis(ms)
    }

    #[test]
    fn pool_key_creation() {
        let key = PoolKey::new("example.com", 8080, true);
        assert_eq!(key.host, "example.com");
        assert_eq!(key.port, 8080);
        assert!(key.is_https);

        let plain_key = PoolKey::http("example.com", None);
        assert_eq!(plain_key.port, 80);
        assert!(!plain_key.is_https);

        let tls_key = PoolKey::https("example.com", None);
        assert_eq!(tls_key.port, 443);
        assert!(tls_key.is_https);
    }

    #[test]
    fn pool_config_builder() {
        let config = PoolConfig::builder()
            .max_connections_per_host(10)
            .max_total_connections(200)
            .idle_timeout(Duration::from_secs(60))
            .build();

        assert_eq!(config.max_connections_per_host, 10);
        assert_eq!(config.max_total_connections, 200);
        assert_eq!(config.idle_timeout, Duration::from_secs(60));
    }

    #[test]
    fn pool_register_and_acquire() {
        let mut pool = Pool::new();
        let key = PoolKey::https("example.com", None);
        let now = make_time(1000);

        // Register a connecting connection
        let id = pool.register_connecting(key.clone(), now, 2);
        assert_eq!(id, 1);

        // Can't acquire while connecting
        assert!(pool.try_acquire(&key, now).is_none());

        // Mark as connected
        assert!(pool.mark_connected(&key, id, now));

        // Now we can acquire it
        let acquired = pool.try_acquire(&key, now);
        assert_eq!(acquired, Some(id));

        // Can't acquire again (it's in use)
        assert!(pool.try_acquire(&key, now).is_none());

        // Release it
        assert!(pool.release(&key, id, now));

        // Can acquire again
        let acquired = pool.try_acquire(&key, now);
        assert_eq!(acquired, Some(id));
    }

    #[test]
    fn pool_connection_limits() {
        let config = PoolConfig::builder()
            .max_connections_per_host(2)
            .max_total_connections(5)
            .build();
        let mut pool = Pool::with_config(config);
        let now = make_time(1000);

        let key1 = PoolKey::https("host1.com", None);
        let key2 = PoolKey::https("host2.com", None);

        // Can create connections up to per-host limit
        assert!(pool.can_create_connection(&key1, now));
        pool.register_connecting(key1.clone(), now, 2);
        assert!(pool.can_create_connection(&key1, now));
        pool.register_connecting(key1.clone(), now, 2);
        assert!(!pool.can_create_connection(&key1, now)); // At limit for host1

        // Can still create for different host
        assert!(pool.can_create_connection(&key2, now));
        pool.register_connecting(key2.clone(), now, 2);
        pool.register_connecting(key2.clone(), now, 2);

        // One more overall
        pool.register_connecting(key2, now, 2);

        // Now at total limit
        let key3 = PoolKey::https("host3.com", None);
        assert!(!pool.can_create_connection(&key3, now));
    }

    #[test]
    fn pool_cleanup_expired() {
        let config = PoolConfig::builder()
            .idle_timeout(Duration::from_millis(100))
            .build();
        let mut pool = Pool::with_config(config);
        let key = PoolKey::https("example.com", None);

        // Create a connection at time 0
        let id = pool.register_connecting(key.clone(), make_time(0), 2);
        assert!(pool.mark_connected(&key, id, make_time(0)));

        // Not expired at time 50
        let removed = pool.cleanup_expired(make_time(50));
        assert_eq!(removed, 0);
        assert!(pool.get_connection_meta(&key, id).is_some());

        // Expired at time 150
        let removed = pool.cleanup_expired(make_time(150));
        assert_eq!(removed, 1);
        assert!(pool.get_connection_meta(&key, id).is_none());
    }

    #[test]
    fn cleanup_expired_entries_returns_removed_connection_ids() {
        let config = PoolConfig::builder()
            .idle_timeout(Duration::from_millis(100))
            .build();
        let mut pool = Pool::with_config(config);
        let key = PoolKey::https("example.com", None);

        let expired_id = pool.register_connecting(key.clone(), make_time(0), 2);
        assert!(pool.mark_connected(&key, expired_id, make_time(0)));
        let live_id = pool.register_connecting(key.clone(), make_time(80), 2);
        assert!(pool.mark_connected(&key, live_id, make_time(80)));

        let removed = pool.cleanup_expired_entries(make_time(150));
        assert_eq!(removed, vec![(key.clone(), expired_id)]);
        assert!(pool.get_connection_meta(&key, expired_id).is_none());
        assert!(pool.get_connection_meta(&key, live_id).is_some());
    }

    #[test]
    fn pool_can_create_connection_ignores_expired_idle() {
        let config = PoolConfig::builder()
            .max_connections_per_host(1)
            .max_total_connections(1)
            .idle_timeout(Duration::from_millis(100))
            .build();
        let mut pool = Pool::with_config(config);
        let key = PoolKey::https("example.com", None);

        let id = pool.register_connecting(key.clone(), make_time(0), 2);
        assert!(pool.mark_connected(&key, id, make_time(0)));

        assert!(
            pool.can_create_connection(&key, make_time(150)),
            "expired idle connection should not block creation"
        );
    }

    #[test]
    fn pool_stats() {
        let mut pool = Pool::new();
        let key = PoolKey::https("example.com", None);
        let now = make_time(1000);

        let id1 = pool.register_connecting(key.clone(), now, 2);
        let id2 = pool.register_connecting(key.clone(), now, 2);

        let stats = pool.stats();
        assert_eq!(stats.total_connections, 2);
        assert_eq!(stats.connecting, 2);
        assert_eq!(stats.idle_connections, 0);
        assert_eq!(stats.connections_created, 2);

        assert!(pool.mark_connected(&key, id1, now));
        assert!(pool.mark_connected(&key, id2, now));

        let stats = pool.stats();
        assert_eq!(stats.idle_connections, 2);
        assert_eq!(stats.connecting, 0);

        pool.try_acquire(&key, now);

        let stats = pool.stats();
        assert_eq!(stats.idle_connections, 1);
        assert_eq!(stats.in_use_connections, 1);
    }

    #[test]
    fn connection_meta_expiry() {
        let mut meta = PooledConnectionMeta::new(1, make_time(0), 2);
        let timeout = Duration::from_millis(100);

        // Not expired while connecting
        assert!(!meta.is_expired(make_time(200), timeout));

        // Mark idle
        meta.mark_idle(make_time(100));

        // Not expired yet
        assert!(!meta.is_expired(make_time(150), timeout));

        // Expired after timeout
        assert!(meta.is_expired(make_time(250), timeout));
    }

    #[test]
    fn remove_nonexistent_is_noop() {
        let mut pool = Pool::new();
        let key = PoolKey::https("example.com", None);
        // Removing from empty pool should not panic
        pool.remove(&key, 999);
        assert_eq!(pool.stats().connections_closed, 0);
    }

    #[test]
    fn release_nonexistent_is_noop() {
        let mut pool = Pool::new();
        let key = PoolKey::https("example.com", None);
        // Releasing a nonexistent connection should not panic
        assert!(!pool.release(&key, 999, make_time(0)));
    }

    #[test]
    fn mark_connected_nonexistent_is_noop() {
        let mut pool = Pool::new();
        let key = PoolKey::https("example.com", None);
        assert!(!pool.mark_connected(&key, 999, make_time(0)));
    }

    #[test]
    fn release_connecting_connection_is_noop() {
        let mut pool = Pool::new();
        let key = PoolKey::https("example.com", None);
        let id = pool.register_connecting(key.clone(), make_time(0), 2);

        assert!(!pool.release(&key, id, make_time(50)));

        let meta = pool
            .get_connection_meta(&key, id)
            .expect("connecting connection should remain present");
        assert_eq!(meta.state, PooledConnectionState::Connecting);
        assert_eq!(meta.last_used, make_time(0));
    }

    #[test]
    fn mark_connected_in_use_connection_is_noop() {
        let mut pool = Pool::new();
        let key = PoolKey::https("example.com", None);
        let id = pool.register_connecting(key.clone(), make_time(0), 2);
        assert!(pool.mark_connected(&key, id, make_time(10)));
        assert_eq!(pool.try_acquire(&key, make_time(20)), Some(id));

        assert!(!pool.mark_connected(&key, id, make_time(50)));

        let meta = pool
            .get_connection_meta(&key, id)
            .expect("in-use connection should remain present");
        assert_eq!(meta.state, PooledConnectionState::InUse);
        assert_eq!(meta.last_used, make_time(10));
        assert_eq!(meta.requests_served, 1);
    }

    #[test]
    fn release_unhealthy_connection_is_noop() {
        let mut meta = PooledConnectionMeta::new(1, make_time(0), 2);
        assert!(meta.mark_connected(make_time(10)));
        meta.mark_unhealthy();

        assert!(!meta.release(make_time(50)));
        assert_eq!(meta.state, PooledConnectionState::Unhealthy);
        assert_eq!(meta.last_used, make_time(10));
    }

    #[test]
    fn pool_default_config() {
        let config = PoolConfig::default();
        assert_eq!(config.max_connections_per_host, 6);
        assert_eq!(config.max_total_connections, 100);
        assert_eq!(config.idle_timeout, Duration::from_secs(90));
        assert_eq!(config.cleanup_interval, Duration::from_secs(30));
    }

    #[test]
    fn acquire_prefers_idle_over_expired() {
        let config = PoolConfig::builder()
            .idle_timeout(Duration::from_millis(100))
            .build();
        let mut pool = Pool::with_config(config);
        let key = PoolKey::https("example.com", None);

        // Create two connections at time 0
        let id1 = pool.register_connecting(key.clone(), make_time(0), 2);
        assert!(pool.mark_connected(&key, id1, make_time(0)));
        let id2 = pool.register_connecting(key.clone(), make_time(50), 2);
        assert!(pool.mark_connected(&key, id2, make_time(50)));

        // At time 120: id1 is expired (idle 120ms), id2 is not (idle 70ms)
        let acquired = pool.try_acquire(&key, make_time(120));
        assert_eq!(acquired, Some(id2));
    }

    #[test]
    fn acquire_uses_lowest_idle_id_for_deterministic_tie_break() {
        let mut pool = Pool::new();
        let key = PoolKey::https("example.com", None);
        let now = make_time(100);

        let id1 = pool.register_connecting(key.clone(), now, 2);
        let id2 = pool.register_connecting(key.clone(), now, 2);
        let id3 = pool.register_connecting(key.clone(), now, 2);

        assert!(pool.mark_connected(&key, id1, now));
        assert!(pool.mark_connected(&key, id2, now));
        assert!(pool.mark_connected(&key, id3, now));

        let acquired = pool.try_acquire(&key, now);
        assert_eq!(acquired, Some(id1));

        let acquired = pool.try_acquire(&key, now);
        assert_eq!(acquired, Some(id2));

        let acquired = pool.try_acquire(&key, now);
        assert_eq!(acquired, Some(id3));
    }

    #[test]
    fn metamorphic_connection_reuse_ignores_ineligible_noise() {
        let config = PoolConfig::builder()
            .idle_timeout(Duration::from_millis(100))
            .build();
        let key = PoolKey::https("example.com", None);
        let now = make_time(100);

        let mut baseline = Pool::with_config(config.clone());
        let baseline_id = baseline.register_connecting(key.clone(), make_time(10), 2);
        assert!(baseline.mark_connected(&key, baseline_id, make_time(10)));
        let baseline_acquired = baseline.try_acquire(&key, now);
        assert_eq!(baseline_acquired, Some(baseline_id));

        let mut transformed = Pool::with_config(config);
        let valid_id = transformed.register_connecting(key.clone(), make_time(10), 2);
        assert!(transformed.mark_connected(&key, valid_id, make_time(10)));

        let expired_id = transformed.register_connecting(key.clone(), make_time(0), 2);
        assert!(transformed.mark_connected(&key, expired_id, make_time(0)));

        let transformed_acquired = transformed.try_acquire(&key, now);
        assert_eq!(
            transformed_acquired, baseline_acquired,
            "expired idle noise must not perturb reuse of the valid idle connection"
        );

        let connecting_noise_id = transformed.register_connecting(key.clone(), make_time(20), 2);
        let transformed_again = transformed.try_acquire(&key, now);
        assert_eq!(
            transformed_again, None,
            "adding only connecting noise must not fabricate reusable capacity after the valid idle connection was consumed"
        );
        let connecting_meta = transformed
            .get_connection_meta(&key, connecting_noise_id)
            .expect("connecting entry remains tracked");
        assert_eq!(connecting_meta.state, PooledConnectionState::Connecting);
    }

    #[test]
    fn metamorphic_target_host_reuse_is_stable_under_unrelated_host_churn() {
        let config = PoolConfig::builder()
            .idle_timeout(Duration::from_millis(100))
            .build();
        let target_key = PoolKey::https("example.com", None);
        let noise_key = PoolKey::https("other.example", None);

        let mut baseline = Pool::with_config(config.clone());
        let baseline_id1 = baseline.register_connecting(target_key.clone(), make_time(10), 2);
        let baseline_id2 = baseline.register_connecting(target_key.clone(), make_time(20), 2);
        assert!(baseline.mark_connected(&target_key, baseline_id1, make_time(10)));
        assert!(baseline.mark_connected(&target_key, baseline_id2, make_time(20)));

        let baseline_first = baseline.try_acquire(&target_key, make_time(50));
        assert_eq!(baseline_first, Some(baseline_id1));
        assert!(baseline.release(&target_key, baseline_id1, make_time(60)));
        let baseline_second = baseline.try_acquire(&target_key, make_time(60));
        assert_eq!(baseline_second, Some(baseline_id1));

        let mut transformed = Pool::with_config(config);
        let transformed_id1 = transformed.register_connecting(target_key.clone(), make_time(10), 2);
        let transformed_id2 = transformed.register_connecting(target_key.clone(), make_time(20), 2);
        assert!(transformed.mark_connected(&target_key, transformed_id1, make_time(10)));
        assert!(transformed.mark_connected(&target_key, transformed_id2, make_time(20)));

        let noise_idle_id = transformed.register_connecting(noise_key.clone(), make_time(5), 2);
        assert!(transformed.mark_connected(&noise_key, noise_idle_id, make_time(5)));
        let noise_connecting_id =
            transformed.register_connecting(noise_key.clone(), make_time(30), 2);

        let transformed_first = transformed.try_acquire(&target_key, make_time(50));
        assert_eq!(
            transformed_first, baseline_first,
            "unrelated host activity must not perturb the first reuse choice for the target host"
        );

        assert!(transformed.release(&target_key, transformed_id1, make_time(60)));
        let transformed_second = transformed.try_acquire(&target_key, make_time(60));
        assert_eq!(
            transformed_second, baseline_second,
            "unrelated host churn must not perturb subsequent reuse ordering for the target host"
        );

        let noise_idle_meta = transformed
            .get_connection_meta(&noise_key, noise_idle_id)
            .expect("idle noise entry remains tracked");
        assert_eq!(noise_idle_meta.state, PooledConnectionState::Idle);

        let noise_connecting_meta = transformed
            .get_connection_meta(&noise_key, noise_connecting_id)
            .expect("connecting noise entry remains tracked");
        assert_eq!(
            noise_connecting_meta.state,
            PooledConnectionState::Connecting
        );
    }

    #[test]
    fn pool_key_debug_clone_eq_ord_hash() {
        use std::collections::HashSet;
        let a = PoolKey::new("example.com", 443, true);
        let b = a.clone();
        assert_eq!(a, b);
        assert_ne!(a, PoolKey::new("other.com", 443, true));
        assert!(a < PoolKey::new("zexample.com", 443, true));
        let dbg = format!("{a:?}");
        assert!(dbg.contains("example.com"));
        assert!(dbg.contains("443"));
        let mut set = HashSet::new();
        set.insert(a);
        assert!(set.contains(&b));
    }

    #[test]
    fn pooled_connection_state_debug_clone_copy_eq() {
        let a = PooledConnectionState::Idle;
        let b = a; // Copy
        let c = a;
        assert_eq!(a, b);
        assert_eq!(a, c);
        assert_ne!(a, PooledConnectionState::InUse);
        assert_ne!(a, PooledConnectionState::Connecting);
        assert_ne!(a, PooledConnectionState::Unhealthy);
        let dbg = format!("{a:?}");
        assert!(dbg.contains("Idle"));
    }
}