armature-core 0.9.0

//! HTTP/1.1 Pipelining Support
//!
//! This module carries a [`PipelineConfig`] alongside connection-level
//! statistics ([`PipelineStats`], [`ConnectionStats`]) and configures hyper's
//! HTTP/1.1 and HTTP/2 connection builders from it. Under the default
//! `h1-backend` feature the HTTP/1.1 half of that is inert: HTTP/1.1 is served
//! by `armature-h1`, which takes its settings from
//! `crate::h1_backend::serve::h1_config` instead, and hyper is left serving
//! HTTP/2 only. Request pipelining itself — multiple requests in flight on one
//! connection without waiting for each response — is a property of the
//! connection driver, not of this module.
//!
//! ## Where each `PipelineConfig` field goes
//!
//! ### Wired on the `h1-backend` path (the default)
//!
//! - `keep_alive_timeout` -> `armature_h1::Limits::idle_timeout`, the deadline
//!   for the next request to begin on an idle keep-alive connection.
//! - `max_header_size` -> `armature_h1::Limits::max_head_bytes`, a byte cap on
//!   the request line plus header section.
//! - `tcp_nodelay` -> `armature_h1::TcpConfig::nodelay`.
//!
//! ### Wired on the hyper path only
//!
//! [`PipelinedHttp1Builder::configure_hyper_builder`] forwards two fields onto
//! hyper's [`hyper::server::conn::http1::Builder`]:
//!
//! - `pipeline_flush` -> [`Builder::pipeline_flush`](hyper::server::conn::http1::Builder::pipeline_flush)
//! - `read_buffer_size` -> [`Builder::max_buf_size`](hyper::server::conn::http1::Builder::max_buf_size)
//!   (clamped up to hyper's documented 8192-byte minimum, since
//!   `PipelineConfig::low_latency()`/`::memory_efficient()` set 4096, which
//!   would otherwise panic)
//!
//! `armature-h1` writes each response as it is produced and grows its read
//! buffer from a fixed chunk size, so it exposes no knob for either. With
//! `h1-backend` on, both reach only the HTTP/2 connections hyper still serves.
//!
//! `tcp_nodelay` is also wired here, but not by this module: on the hyper path
//! the caller (`Application`) calls `TcpStream::set_nodelay` on the accepted
//! socket itself.
//!
//! ### Wired on neither path
//!
//! `mode`, `max_concurrent`, `max_buffered_requests`,
//! `max_requests_per_connection` and `write_buffer_size` are recorded on the
//! config (and `mode` is echoed into a `tracing` log line) but nothing reads
//! them. The reasons differ per backend:
//!
//! - On the `h1-backend` path the three concurrency/lifecycle limits are
//!   structurally moot: `armature-h1` reads no further than one complete head
//!   and does not read again until that request's response has been written,
//!   so there is never more than one request in flight on a connection to
//!   limit. `mode` and `write_buffer_size` have nothing to attach to either.
//! - On the hyper path, hyper's per-connection H1 builder has no API surface
//!   for request scheduling, per-connection request counts, or a separate
//!   write-buffer size (only the combined `max_buf_size`, which
//!   `read_buffer_size` already maps to). Implementing them would need custom
//!   logic layered above hyper that this crate does not have.
//!
//! Selecting a `PipelineMode` therefore has **no effect** on server behavior
//! beyond appearing in logs and being queryable via
//! [`PipelineMode::maintains_order`] / [`PipelineMode::is_concurrent`].
//!
//! ## Configuration
//!
//! ```rust,ignore
//! use armature_core::pipeline::{PipelineConfig, PipelineMode};
//!
//! let config = PipelineConfig::builder()
//!     .mode(PipelineMode::Concurrent) // informational only
//!     .max_concurrent(16)             // unwired on both backends
//!     .pipeline_flush(true)           // hyper path only
//!     .keep_alive_timeout(Duration::from_secs(60)) // h1-backend path only
//!     .build();
//! ```

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::time::Duration;

/// Pipeline processing mode
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum PipelineMode {
    /// Process requests strictly in order (HTTP/1.1 compliant)
    /// Responses are sent in the same order as requests
    #[default]
    Sequential,

    /// Process requests concurrently but respond in order
    /// Better throughput while maintaining HTTP/1.1 compliance
    Concurrent,

    /// Process and respond as fast as possible (out-of-order)
    /// Not HTTP/1.1 compliant but maximum throughput
    /// Only use with clients that support out-of-order responses
    OutOfOrder,
}

impl PipelineMode {
    /// Check if this mode maintains response ordering
    #[inline]
    pub fn maintains_order(&self) -> bool {
        matches!(self, Self::Sequential | Self::Concurrent)
    }

    /// Check if this mode allows concurrent processing
    #[inline]
    pub fn is_concurrent(&self) -> bool {
        matches!(self, Self::Concurrent | Self::OutOfOrder)
    }
}

/// Configuration for HTTP/1.1 pipelining
///
/// Which of these fields reach server behavior depends on the backend: see
/// the [module docs](self) for the per-backend split, and each field below
/// for where it lands.
#[derive(Debug, Clone)]
pub struct PipelineConfig {
    /// Pipeline processing mode.
    ///
    /// **Not wired on either path**: nothing branches on this value; it is
    /// only echoed into a `tracing` log line by the caller. Neither backend
    /// exposes a request-scheduling policy for it to select.
    pub mode: PipelineMode,

    /// Maximum number of concurrent requests per connection.
    ///
    /// **Not wired on either path.** On the `h1-backend` path — the default —
    /// the limit is structurally moot: `armature-h1` reads no further than one
    /// complete head and does not read again until that request's response has
    /// been written, so there is never more than one request in flight on a
    /// connection to limit. On the hyper path, hyper's H1 builder has no
    /// per-connection concurrency-limit knob, and enforcing one would require
    /// custom scheduling logic above hyper that does not exist.
    pub max_concurrent: usize,

    /// Enable pipeline flush optimization.
    /// When true, responses are flushed in batches for better I/O efficiency.
    ///
    /// **Wired**: forwarded to
    /// [`hyper::server::conn::http1::Builder::pipeline_flush`].
    ///
    /// Applies only on the hyper path. With the `h1-backend` feature on — the
    /// default — HTTP/1.1 is served by `armature-h1`, which writes each
    /// response as it is produced and has no batching knob to attach this to;
    /// the field then reaches only the HTTP/2 connections hyper still serves.
    pub pipeline_flush: bool,

    /// Maximum number of pipelined requests to buffer.
    ///
    /// **Not wired on either path.** On the `h1-backend` path there is nothing
    /// to buffer: `armature-h1` reads no further than one complete head and
    /// does not read again until that request's response has been written, so
    /// a second pipelined request is never taken off the socket to be queued.
    /// On the hyper path, hyper's H1 builder has no request-buffering limit to
    /// apply this to.
    pub max_buffered_requests: usize,

    /// Keep-alive timeout for idle connections.
    ///
    /// **Wired on the `h1-backend` path**, to `armature-h1`'s
    /// `Limits::idle_timeout` — the deadline for the next request to begin on
    /// an idle keep-alive connection, which is what this field always meant.
    ///
    /// Not wired on the hyper path: hyper's H1 builder only exposes
    /// `keep_alive(bool)` (used, hardcoded `true`) and a header-read timeout
    /// with different semantics that requires a [`hyper::rt::Timer`] this
    /// crate does not currently supply.
    pub keep_alive_timeout: Duration,

    /// How long a handler may run before the connection answers `408` and
    /// closes, or `None` for no limit.
    ///
    /// **Wired on the `h1-backend` path**, to `armature-h1`'s
    /// `Limits::body_timeout` — which despite its name races the *whole*
    /// handler future, and so covers the handler's reads of the request body
    /// as well as its own work.
    ///
    /// `None` by default, and deliberately: the hyper path supplies no
    /// [`hyper::rt::Timer`] and therefore imposes no request deadline at all,
    /// so defaulting this to a finite value would make a backend swap silently
    /// cancel every long-poll, SSE stream and slow upload the previous
    /// behaviour allowed. Set it when you know your handlers' upper bound —
    /// a request deadline is worth having, but it has to be one you asked for.
    ///
    /// Note what this does *not* bound: a peer that has sent a complete head
    /// and then trickles its body. That is the same deadline, so a deployment
    /// exposed directly to the internet rather than behind a proxy that
    /// enforces its own request timeout should set this.
    pub request_timeout: Option<Duration>,

    /// How long a single response write may take before the connection is torn
    /// down, or `None` for no limit.
    ///
    /// **Wired on the `h1-backend` path**, to `armature-h1`'s
    /// `Limits::write_timeout`.
    ///
    /// Unlike [`request_timeout`](Self::request_timeout) this defaults to a
    /// finite value, because the failure it prevents is not a slow handler but
    /// a peer that stops reading: `armature-h1` caps neither connection count
    /// nor write duration, so an unbounded write deadline lets a client hold a
    /// worker's connection slot, its file descriptor and its full response
    /// buffer indefinitely by shrinking its receive window to zero. Five
    /// minutes is generous enough for a large response over a slow link and
    /// finite enough that the hold is not free.
    pub write_timeout: Option<Duration>,

    /// Maximum requests per connection before forcing close.
    /// Helps prevent resource exhaustion.
    ///
    /// **Not wired on either path.** `armature-h1` serves one request at a
    /// time per connection and exposes no per-connection request budget, and
    /// hyper's H1 builder has no request-count limit either; enforcing this
    /// would require closing the connection from caller-side logic after N
    /// requests, which neither backend gives this crate a hook for.
    pub max_requests_per_connection: Option<u64>,

    /// Enable TCP_NODELAY for lower latency.
    ///
    /// **Wired on both paths**, but not by this module — and by a different
    /// mechanism on each. On the hyper path the caller (`Application`) reads
    /// this field directly and calls `TcpStream::set_nodelay` on the accepted
    /// socket before handing it to hyper. On the `h1-backend` path — the
    /// default — this process does not own the accepted socket, so the value
    /// is forwarded as `armature_h1::TcpConfig::nodelay` and applied by
    /// `armature-h1`'s own accept loop.
    pub tcp_nodelay: bool,

    /// Read buffer size hint (bytes).
    ///
    /// **Wired**: forwarded to
    /// [`hyper::server::conn::http1::Builder::max_buf_size`].
    ///
    /// Applies only on the hyper path. With the `h1-backend` feature on — the
    /// default — HTTP/1.1 is served by `armature-h1`, which grows its read
    /// buffer from a fixed chunk size and takes no size hint; the field then
    /// reaches only the HTTP/2 connections hyper still serves.
    pub read_buffer_size: usize,

    /// Write buffer size hint (bytes).
    ///
    /// **Not wired on either path**: `armature-h1` writes each response as it
    /// is produced and takes no write-buffer hint, and hyper's H1 builder has
    /// no separate write-buffer-size knob (only the combined `max_buf_size`,
    /// which `read_buffer_size` already maps to).
    pub write_buffer_size: usize,

    /// Maximum header size (bytes).
    ///
    /// **Wired on the `h1-backend` path**, to `armature-h1`'s
    /// `Limits::max_head_bytes` — a byte cap on the request line plus header
    /// section, which is what this field always meant.
    ///
    /// Not wired on the hyper path: hyper's H1 builder exposes
    /// [`hyper::server::conn::http1::Builder::max_headers`], but that
    /// method limits a *count* of headers (default 100), not a byte size,
    /// so this field cannot be applied to it without changing its meaning.
    pub max_header_size: usize,
}

impl Default for PipelineConfig {
    fn default() -> Self {
        Self {
            mode: PipelineMode::Concurrent,
            max_concurrent: 16,
            pipeline_flush: true,
            max_buffered_requests: 64,
            keep_alive_timeout: Duration::from_secs(60),
            // No handler deadline unless asked for; a finite write deadline
            // because a stalled reader is not a legitimate slow handler. See
            // each field's documentation for why the two differ.
            request_timeout: None,
            write_timeout: Some(Duration::from_secs(300)),
            max_requests_per_connection: Some(10_000),
            tcp_nodelay: true,
            read_buffer_size: 8192,
            write_buffer_size: 8192,
            max_header_size: 16384,
        }
    }
}

impl PipelineConfig {
    /// Create a new builder for PipelineConfig
    pub fn builder() -> PipelineConfigBuilder {
        PipelineConfigBuilder::default()
    }

    /// Create a high-performance configuration
    pub fn high_performance() -> Self {
        Self {
            mode: PipelineMode::Concurrent,
            max_concurrent: 32,
            pipeline_flush: true,
            max_buffered_requests: 128,
            keep_alive_timeout: Duration::from_secs(120),
            request_timeout: None,
            write_timeout: Some(Duration::from_secs(300)),
            max_requests_per_connection: Some(100_000),
            tcp_nodelay: true,
            read_buffer_size: 16384,
            write_buffer_size: 16384,
            max_header_size: 32768,
        }
    }

    /// Create a low-latency configuration
    pub fn low_latency() -> Self {
        Self {
            mode: PipelineMode::Sequential,
            max_concurrent: 1,
            pipeline_flush: false,
            max_buffered_requests: 16,
            keep_alive_timeout: Duration::from_secs(30),
            request_timeout: None,
            // Tighter than the default: a configuration asking for low latency
            // is not one that expects a client to spend five minutes reading.
            write_timeout: Some(Duration::from_secs(60)),
            max_requests_per_connection: Some(1000),
            tcp_nodelay: true,
            read_buffer_size: 4096,
            write_buffer_size: 4096,
            max_header_size: 8192,
        }
    }

    /// Create a memory-efficient configuration
    pub fn memory_efficient() -> Self {
        Self {
            mode: PipelineMode::Sequential,
            max_concurrent: 4,
            pipeline_flush: true,
            max_buffered_requests: 32,
            keep_alive_timeout: Duration::from_secs(30),
            request_timeout: None,
            write_timeout: Some(Duration::from_secs(300)),
            max_requests_per_connection: Some(1000),
            tcp_nodelay: false,
            read_buffer_size: 4096,
            write_buffer_size: 4096,
            max_header_size: 8192,
        }
    }
}

/// Builder for PipelineConfig
#[derive(Debug, Clone, Default)]
pub struct PipelineConfigBuilder {
    config: PipelineConfig,
}

impl PipelineConfigBuilder {
    /// Set the pipeline processing mode
    pub fn mode(mut self, mode: PipelineMode) -> Self {
        self.config.mode = mode;
        self
    }

    /// Set maximum concurrent requests
    pub fn max_concurrent(mut self, max: usize) -> Self {
        self.config.max_concurrent = max;
        self
    }

    /// Enable or disable pipeline flush optimization
    pub fn pipeline_flush(mut self, enable: bool) -> Self {
        self.config.pipeline_flush = enable;
        self
    }

    /// Set maximum buffered requests
    pub fn max_buffered_requests(mut self, max: usize) -> Self {
        self.config.max_buffered_requests = max;
        self
    }

    /// Set keep-alive timeout
    pub fn keep_alive_timeout(mut self, timeout: Duration) -> Self {
        self.config.keep_alive_timeout = timeout;
        self
    }

    /// Set maximum requests per connection
    pub fn max_requests_per_connection(mut self, max: Option<u64>) -> Self {
        self.config.max_requests_per_connection = max;
        self
    }

    /// Enable or disable TCP_NODELAY
    pub fn tcp_nodelay(mut self, enable: bool) -> Self {
        self.config.tcp_nodelay = enable;
        self
    }

    /// Set read buffer size
    pub fn read_buffer_size(mut self, size: usize) -> Self {
        self.config.read_buffer_size = size;
        self
    }

    /// Set write buffer size
    pub fn write_buffer_size(mut self, size: usize) -> Self {
        self.config.write_buffer_size = size;
        self
    }

    /// Set maximum header size
    pub fn max_header_size(mut self, size: usize) -> Self {
        self.config.max_header_size = size;
        self
    }

    /// Build the configuration
    pub fn build(self) -> PipelineConfig {
        self.config
    }
}

// ============================================================================
// Connection Statistics
// ============================================================================

/// Statistics for a pipelined connection
#[derive(Debug)]
pub struct ConnectionStats {
    /// Total requests processed
    requests_processed: AtomicU64,
    /// Currently pending requests
    pending_requests: AtomicUsize,
    /// Total bytes received
    bytes_received: AtomicU64,
    /// Total bytes sent
    bytes_sent: AtomicU64,
    /// Pipeline depth (how many requests are queued)
    pipeline_depth: AtomicUsize,
}

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

impl ConnectionStats {
    /// Create new connection statistics
    pub fn new() -> Self {
        Self {
            requests_processed: AtomicU64::new(0),
            pending_requests: AtomicUsize::new(0),
            bytes_received: AtomicU64::new(0),
            bytes_sent: AtomicU64::new(0),
            pipeline_depth: AtomicUsize::new(0),
        }
    }

    /// Record a request received
    #[inline]
    pub fn request_received(&self, bytes: u64) {
        self.pending_requests.fetch_add(1, Ordering::Relaxed);
        self.bytes_received.fetch_add(bytes, Ordering::Relaxed);
        self.pipeline_depth.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a response sent
    #[inline]
    pub fn response_sent(&self, bytes: u64) {
        self.requests_processed.fetch_add(1, Ordering::Relaxed);
        self.pending_requests.fetch_sub(1, Ordering::Relaxed);
        self.bytes_sent.fetch_add(bytes, Ordering::Relaxed);
        self.pipeline_depth.fetch_sub(1, Ordering::Relaxed);
    }

    /// Get total requests processed
    #[inline]
    pub fn requests_processed(&self) -> u64 {
        self.requests_processed.load(Ordering::Relaxed)
    }

    /// Get currently pending requests
    #[inline]
    pub fn pending_requests(&self) -> usize {
        self.pending_requests.load(Ordering::Relaxed)
    }

    /// Get current pipeline depth
    #[inline]
    pub fn pipeline_depth(&self) -> usize {
        self.pipeline_depth.load(Ordering::Relaxed)
    }

    /// Get total bytes received
    #[inline]
    pub fn bytes_received(&self) -> u64 {
        self.bytes_received.load(Ordering::Relaxed)
    }

    /// Get total bytes sent
    #[inline]
    pub fn bytes_sent(&self) -> u64 {
        self.bytes_sent.load(Ordering::Relaxed)
    }
}

// ============================================================================
// Global Pipeline Statistics
// ============================================================================

/// Global statistics for all pipelined connections
#[derive(Debug, Default)]
pub struct PipelineStats {
    /// Active connections
    active_connections: AtomicUsize,
    /// Total connections ever
    total_connections: AtomicU64,
    /// Total requests processed
    total_requests: AtomicU64,
    /// Average pipeline depth (running average * 100 for precision)
    avg_pipeline_depth: AtomicU64,
    /// Maximum pipeline depth seen
    max_pipeline_depth: AtomicUsize,
}

impl PipelineStats {
    /// Create new global pipeline statistics
    pub fn new() -> Self {
        Self::default()
    }

    /// Record a new connection
    #[inline]
    pub fn connection_opened(&self) {
        self.active_connections.fetch_add(1, Ordering::Relaxed);
        self.total_connections.fetch_add(1, Ordering::Relaxed);
    }

    /// Record a connection closed
    #[inline]
    pub fn connection_closed(&self) {
        self.active_connections.fetch_sub(1, Ordering::Relaxed);
    }

    /// Record a request processed
    #[inline]
    pub fn request_processed(&self) {
        self.total_requests.fetch_add(1, Ordering::Relaxed);
    }

    /// Update pipeline depth statistics
    #[inline]
    pub fn update_pipeline_depth(&self, depth: usize) {
        // Update max depth
        self.max_pipeline_depth.fetch_max(depth, Ordering::Relaxed);

        // Update running average (simplified EMA)
        let current = self.avg_pipeline_depth.load(Ordering::Relaxed);
        let new_avg = (current * 95 + (depth as u64 * 100) * 5) / 100;
        self.avg_pipeline_depth.store(new_avg, Ordering::Relaxed);
    }

    /// Get active connections
    #[inline]
    pub fn active_connections(&self) -> usize {
        self.active_connections.load(Ordering::Relaxed)
    }

    /// Get total connections
    #[inline]
    pub fn total_connections(&self) -> u64 {
        self.total_connections.load(Ordering::Relaxed)
    }

    /// Get total requests
    #[inline]
    pub fn total_requests(&self) -> u64 {
        self.total_requests.load(Ordering::Relaxed)
    }

    /// Get average pipeline depth
    #[inline]
    pub fn avg_pipeline_depth(&self) -> f64 {
        self.avg_pipeline_depth.load(Ordering::Relaxed) as f64 / 100.0
    }

    /// Get maximum pipeline depth
    #[inline]
    pub fn max_pipeline_depth(&self) -> usize {
        self.max_pipeline_depth.load(Ordering::Relaxed)
    }
}

// ============================================================================
// Pipeline-aware HTTP/1.1 Connection Handler
// ============================================================================

/// A wrapper for configuring Hyper's http1 builder with pipelining options
pub struct PipelinedHttp1Builder {
    config: PipelineConfig,
    stats: Arc<PipelineStats>,
}

impl PipelinedHttp1Builder {
    /// Create a new pipelined HTTP/1.1 builder
    pub fn new(config: PipelineConfig) -> Self {
        Self {
            config,
            stats: Arc::new(PipelineStats::new()),
        }
    }

    /// Create with shared statistics
    pub fn with_stats(config: PipelineConfig, stats: Arc<PipelineStats>) -> Self {
        Self { config, stats }
    }

    /// Get the configuration
    pub fn config(&self) -> &PipelineConfig {
        &self.config
    }

    /// Get shared statistics
    pub fn stats(&self) -> Arc<PipelineStats> {
        Arc::clone(&self.stats)
    }

    /// Configure a hyper `http1::Builder` from this instance's [`PipelineConfig`].
    ///
    /// Only `pipeline_flush` and `read_buffer_size` are actually applied here;
    /// see the [module docs](self) for why the remaining `PipelineConfig`
    /// fields (`mode`, `max_concurrent`, `max_buffered_requests`,
    /// `keep_alive_timeout`, `max_requests_per_connection`,
    /// `write_buffer_size`, `max_header_size`) are not wired through hyper's
    /// H1 builder.
    #[inline]
    pub fn configure_hyper_builder(&self) -> hyper::server::conn::http1::Builder {
        let mut builder = hyper::server::conn::http1::Builder::new();

        // Enable pipeline flush for batched response sending
        builder.pipeline_flush(self.config.pipeline_flush);

        // Set maximum buffer sizes. hyper's `max_buf_size` panics below its
        // documented minimum (8192 bytes); `PipelineConfig::low_latency()` and
        // `::memory_efficient()` set `read_buffer_size: 4096`, so clamp up to
        // avoid a panic while still honoring larger configured values.
        builder.max_buf_size(self.config.read_buffer_size.max(8192));

        // Preserve header case (for compatibility)
        builder.preserve_header_case(true);

        // Keep-alive is essential for pipelining
        builder.keep_alive(true);

        builder
    }
}

// ============================================================================
// Tests
// ============================================================================

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

    #[test]
    fn test_pipeline_mode_properties() {
        assert!(PipelineMode::Sequential.maintains_order());
        assert!(PipelineMode::Concurrent.maintains_order());
        assert!(!PipelineMode::OutOfOrder.maintains_order());

        assert!(!PipelineMode::Sequential.is_concurrent());
        assert!(PipelineMode::Concurrent.is_concurrent());
        assert!(PipelineMode::OutOfOrder.is_concurrent());
    }

    #[test]
    fn test_config_builder() {
        let config = PipelineConfig::builder()
            .mode(PipelineMode::Concurrent)
            .max_concurrent(32)
            .pipeline_flush(true)
            .keep_alive_timeout(Duration::from_secs(120))
            .build();

        assert_eq!(config.mode, PipelineMode::Concurrent);
        assert_eq!(config.max_concurrent, 32);
        assert!(config.pipeline_flush);
        assert_eq!(config.keep_alive_timeout, Duration::from_secs(120));
    }

    #[test]
    fn test_high_performance_config() {
        let config = PipelineConfig::high_performance();
        assert_eq!(config.mode, PipelineMode::Concurrent);
        assert_eq!(config.max_concurrent, 32);
        assert!(config.pipeline_flush);
    }

    #[test]
    fn test_low_latency_config() {
        let config = PipelineConfig::low_latency();
        assert_eq!(config.mode, PipelineMode::Sequential);
        assert_eq!(config.max_concurrent, 1);
        assert!(!config.pipeline_flush);
    }

    #[test]
    fn test_connection_stats() {
        let stats = ConnectionStats::new();

        stats.request_received(100);
        assert_eq!(stats.pending_requests(), 1);
        assert_eq!(stats.pipeline_depth(), 1);
        assert_eq!(stats.bytes_received(), 100);

        stats.request_received(200);
        assert_eq!(stats.pending_requests(), 2);
        assert_eq!(stats.pipeline_depth(), 2);

        stats.response_sent(150);
        assert_eq!(stats.pending_requests(), 1);
        assert_eq!(stats.requests_processed(), 1);
        assert_eq!(stats.bytes_sent(), 150);
    }

    #[test]
    fn test_global_pipeline_stats() {
        let stats = PipelineStats::new();

        stats.connection_opened();
        stats.connection_opened();
        assert_eq!(stats.active_connections(), 2);
        assert_eq!(stats.total_connections(), 2);

        stats.connection_closed();
        assert_eq!(stats.active_connections(), 1);
        assert_eq!(stats.total_connections(), 2);

        stats.request_processed();
        stats.request_processed();
        assert_eq!(stats.total_requests(), 2);

        stats.update_pipeline_depth(5);
        stats.update_pipeline_depth(10);
        assert_eq!(stats.max_pipeline_depth(), 10);
    }

    #[test]
    fn test_pipelined_builder() {
        let config = PipelineConfig::default();
        let builder = PipelinedHttp1Builder::new(config);

        assert_eq!(builder.config().mode, PipelineMode::Concurrent);
        assert_eq!(builder.stats().active_connections(), 0);

        // Configure Hyper builder
        let _hyper_builder = builder.configure_hyper_builder();
    }
}