diff --git a/Sources/NIOHTTPServer/Configuration/NIOHTTPServer+SwiftConfiguration.swift b/Sources/NIOHTTPServer/Configuration/NIOHTTPServer+SwiftConfiguration.swift
index ce55ff8..807ef45 100644
--- a/Sources/NIOHTTPServer/Configuration/NIOHTTPServer+SwiftConfiguration.swift
+++ b/Sources/NIOHTTPServer/Configuration/NIOHTTPServer+SwiftConfiguration.swift
@@ -73,9 +73,11 @@ extension NIOHTTPServerConfiguration {
             transportSecurity: try .init(
                 config: snapshot.scoped(to: "transportSecurity"),
                 customCertificateVerificationCallback: customCertificateVerificationCallback
-            ),
-            backpressureStrategy: .init(config: snapshot.scoped(to: "backpressureStrategy"))
+            )
         )
+        self.backpressureStrategy = .init(config: snapshot.scoped(to: "backpressureStrategy"))
+        self.maxConnections = snapshot.int(forKey: "maxConnections")
+        self.connectionTimeouts = .init(config: snapshot.scoped(to: "connectionTimeouts"))
     }
 
     /// Reads bind targets from either the singular `bindTarget` scope or the plural `bindTargets` scope.
@@ -483,4 +485,25 @@ extension CertificateVerificationMode {
         }
     }
 }
+@available(anyAppleOS 26.0, *)
+extension NIOHTTPServerConfiguration.ConnectionTimeouts {
+    /// Initialize connection timeouts configuration from a config reader.
+    ///
+    /// ## Configuration keys:
+    /// - `idle` (int, optional, default: nil): Maximum time in seconds a connection can remain idle.
+    /// - `readHeader` (int, optional, default: nil): Maximum time in seconds to receive request headers
+    /// after a connection is established.
+    /// - `readBody` (int, optional, default: nil): Maximum time in seconds to receive the complete request
+    /// body after headers have been received.
+    ///
+    /// - Parameter config: The configuration reader.
+    public init(config: ConfigSnapshotReader) {
+        self.init(
+            idle: config.int(forKey: "idle").map { .seconds($0) },
+            readHeader: config.int(forKey: "readHeader").map { .seconds($0) },
+            readBody: config.int(forKey: "readBody").map { .seconds($0) }
+        )
+    }
+}
+
 #endif  // Configuration
diff --git a/Sources/NIOHTTPServer/Configuration/NIOHTTPServerConfiguration.swift b/Sources/NIOHTTPServer/Configuration/NIOHTTPServerConfiguration.swift
index 5cf6e05..2f462e3 100644
--- a/Sources/NIOHTTPServer/Configuration/NIOHTTPServerConfiguration.swift
+++ b/Sources/NIOHTTPServer/Configuration/NIOHTTPServerConfiguration.swift
@@ -230,6 +230,60 @@ public struct NIOHTTPServerConfiguration: Sendable {
         }
     }
 
+    /// Configuration for connection timeouts.
+    ///
+    /// Timeouts are enabled by default with reasonable values to protect against
+    /// slow or idle connections. Individual timeouts can be disabled by setting
+    /// them to `nil`.
+    public struct ConnectionTimeouts: Sendable {
+        /// Maximum time the connection may sit with no request in flight before being closed.
+        ///
+        /// On HTTP/1.1, the timer runs between requests on a keep-alive connection: it starts
+        /// when the connection becomes active and is rescheduled after each response `.end` is
+        /// written. The timer is cancelled when an inbound request `.head` is observed.
+        ///
+        /// On HTTP/2, this is delegated to `NIOHTTP2ServerConnectionManagementHandler`'s
+        /// `maxIdleTime`, which fires when no streams have been open for the configured duration
+        /// and triggers a graceful shutdown.
+        ///
+        /// `nil` means no idle timeout.
+        public var idle: Duration?
+
+        /// Maximum time allowed to receive the complete request headers
+        /// after a connection is established. `nil` means no timeout.
+        public var readHeader: Duration?
+
+        /// Maximum time allowed to receive the complete request body
+        /// after headers have been received. `nil` means no timeout.
+        public var readBody: Duration?
+
+        /// - Parameters:
+        ///   - idle: Maximum idle time before the connection is closed.
+        ///   - readHeader: Maximum time to receive request headers after a connection is established.
+        ///   - readBody: Maximum time to receive the complete request body after headers have been received.
+        public init(
+            idle: Duration? = Self.defaultIdle,
+            readHeader: Duration? = Self.defaultReadHeader,
+            readBody: Duration? = Self.defaultReadBody
+        ) {
+            self.idle = idle
+            self.readHeader = readHeader
+            self.readBody = readBody
+        }
+
+        @inlinable
+        static var defaultIdle: Duration? { .seconds(60) }
+
+        @inlinable
+        static var defaultReadHeader: Duration? { .seconds(30) }
+
+        @inlinable
+        static var defaultReadBody: Duration? { .seconds(60) }
+
+        /// Default timeout values: 60s idle, 30s read header, 60s read body.
+        public static var defaults: Self { .init() }
+    }
+
     /// Network binding configuration specifying all addresses where the server should listen.
     public var bindTargets: [BindTarget]
 
@@ -242,18 +296,37 @@ public struct NIOHTTPServerConfiguration: Sendable {
     /// Backpressure strategy to use in the server.
     public var backpressureStrategy: BackPressureStrategy
 
+    /// The maximum number of concurrent connections the server will accept.
+    ///
+    /// When this limit is reached, the server stops accepting new connections
+    /// until existing ones close. `nil` means unlimited (the default).
+    ///
+    /// - Precondition: Must be greater than 0 if non-`nil`.
+    public var maxConnections: Int? {
+        didSet {
+            if let maxConnections, maxConnections <= 0 {
+                preconditionFailure("`maxConnections` must be greater than 0.")
+            }
+        }
+    }
+
+    /// Configuration for connection timeouts.
+    public var connectionTimeouts: ConnectionTimeouts
+
     /// Create a new configuration with multiple bind targets.
+    ///
+    /// Other configuration properties (``backpressureStrategy``, ``maxConnections``,
+    /// ``connectionTimeouts``) are initialized to their defaults and can be set on the resulting
+    /// value before passing it to ``NIOHTTPServer``.
+    ///
     /// - Parameters:
     ///   - bindTargets: An array of ``BindTarget`` values specifying where the server should listen.
     ///   - supportedHTTPVersions: The HTTP protocol versions the server should support.
     ///   - transportSecurity: The transport security mode (plaintext, TLS, or mTLS).
-    ///   - backpressureStrategy: A ``BackPressureStrategy``.
-    ///   Defaults to ``BackPressureStrategy/watermark(low:high:)`` with a low watermark of 2 and a high of 10.
     public init(
         bindTargets: [BindTarget],
         supportedHTTPVersions: Set<HTTPVersion>,
-        transportSecurity: TransportSecurity,
-        backpressureStrategy: BackPressureStrategy = .defaults
+        transportSecurity: TransportSecurity
     ) throws {
         if bindTargets.isEmpty {
             throw NIOHTTPServerConfigurationError.noBindTargetsSpecified
@@ -274,27 +347,30 @@ public struct NIOHTTPServerConfiguration: Sendable {
         self.bindTargets = bindTargets
         self.supportedHTTPVersions = supportedHTTPVersions
         self.transportSecurity = transportSecurity
-        self.backpressureStrategy = backpressureStrategy
+        self.backpressureStrategy = .defaults
+        self.maxConnections = nil
+        self.connectionTimeouts = .defaults
     }
 
     /// Create a new configuration with a single bind target.
+    ///
+    /// Other configuration properties (``backpressureStrategy``, ``maxConnections``,
+    /// ``connectionTimeouts``) are initialized to their defaults and can be set on the resulting
+    /// value before passing it to ``NIOHTTPServer``.
+    ///
     /// - Parameters:
     ///   - bindTarget: A ``BindTarget``.
     ///   - supportedHTTPVersions: The HTTP protocol versions the server should support.
     ///   - transportSecurity: The transport security mode (plaintext, TLS, or mTLS).
-    ///   - backpressureStrategy: A ``BackPressureStrategy``.
-    ///   Defaults to ``BackPressureStrategy/watermark(low:high:)`` with a low watermark of 2 and a high of 10.
     public init(
         bindTarget: BindTarget,
         supportedHTTPVersions: Set<HTTPVersion>,
-        transportSecurity: TransportSecurity,
-        backpressureStrategy: BackPressureStrategy = .defaults
+        transportSecurity: TransportSecurity
     ) throws {
         try self.init(
             bindTargets: [bindTarget],
             supportedHTTPVersions: supportedHTTPVersions,
-            transportSecurity: transportSecurity,
-            backpressureStrategy: backpressureStrategy
+            transportSecurity: transportSecurity
         )
     }
 }
diff --git a/Sources/NIOHTTPServer/ConnectionLimitHandler.swift b/Sources/NIOHTTPServer/ConnectionLimitHandler.swift
new file mode 100644
index 0000000..afb47f2
--- /dev/null
+++ b/Sources/NIOHTTPServer/ConnectionLimitHandler.swift
@@ -0,0 +1,66 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift HTTP Server open source project
+//
+// Copyright (c) 2026 Apple Inc. and the Swift HTTP Server project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of Swift HTTP Server project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+
+/// A channel handler installed on the server (parent) channel that limits the
+/// number of concurrent connections by gating `read()` calls.
+///
+/// When the number of active connections reaches `maxConnections`, this handler
+/// stops forwarding `read()` events, which prevents NIO from calling `accept()`
+/// on the listening socket. When a connection closes and count drops below the
+/// limit, `read()` is re-triggered to resume accepting.
+final class ConnectionLimitHandler: ChannelDuplexHandler {
+    typealias InboundIn = Channel
+    typealias InboundOut = Channel
+    typealias OutboundIn = Channel
+
+    private let maxConnections: Int
+    private var activeConnections: Int = 0
+    private var pendingRead: Bool = false
+
+    init(maxConnections: Int) {
+        self.maxConnections = maxConnections
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        let childChannel = self.unwrapInboundIn(data)
+        self.activeConnections += 1
+
+        let loopBoundSelf = NIOLoopBound(self, eventLoop: context.eventLoop)
+        let loopBoundContext = NIOLoopBound(context, eventLoop: context.eventLoop)
+        let eventLoop = context.eventLoop
+        childChannel.closeFuture.whenComplete { _ in
+            eventLoop.execute {
+                let `self` = loopBoundSelf.value
+                let context = loopBoundContext.value
+                `self`.activeConnections -= 1
+                if `self`.pendingRead && `self`.activeConnections <= `self`.maxConnections {
+                    `self`.pendingRead = false
+                    context.read()
+                }
+            }
+        }
+
+        context.fireChannelRead(data)
+    }
+
+    func read(context: ChannelHandlerContext) {
+        if self.activeConnections <= self.maxConnections {
+            context.read()
+        } else {
+            self.pendingRead = true
+        }
+    }
+}
diff --git a/Sources/NIOHTTPServer/Documentation.docc/SwiftConfigurationIntegration.md b/Sources/NIOHTTPServer/Documentation.docc/SwiftConfigurationIntegration.md
index 772ca54..77be261 100644
--- a/Sources/NIOHTTPServer/Documentation.docc/SwiftConfigurationIntegration.md
+++ b/Sources/NIOHTTPServer/Documentation.docc/SwiftConfigurationIntegration.md
@@ -29,8 +29,8 @@ let serverConfiguration = try NIOHTTPServerConfiguration(config: config)
 
 ### Configuration key reference
 
-``NIOHTTPServerConfiguration`` is comprised of four components. Provide the configuration for each component under its
-respective key prefix.
+``NIOHTTPServerConfiguration`` is comprised of several components. Provide the configuration for each component under
+its respective key prefix.
 
 > Important: Exactly one of `bindTarget` (singular, for a single address) or `bindTargets` (plural, for multiple
 > addresses) must be provided. Providing both results in an error.
@@ -62,6 +62,10 @@ respective key prefix.
 |                               | `certificateVerificationMode`     | `string`       | Required for `"mTLS"`, permitted values: `"optionalVerification"`, `"noHostnameVerification"`                                 | -       |
 | `backpressureStrategy`        | `lowWatermark`                    | `int`          | Optional                                                                                                                      | 2       |
 |                               | `highWatermark`                   | `int`          | Optional                                                                                                                      | 10      |
+|                               | `maxConnections`                  | `int`          | Optional                                                                                                                      | nil     |
+| `connectionTimeouts`          | `idle`                            | `int`          | Optional                                                                                                                      | nil     |
+|                               | `readHeader`                      | `int`          | Optional                                                                                                                      | nil     |
+|                               | `readBody`                        | `int`          | Optional                                                                                                                      | nil     |
 
 
 The `credentialSource` determines how server credentials are provided:
@@ -113,6 +117,12 @@ key were omitted.
     "backpressureStrategy": {
         "lowWatermark": 2,                  // default: 2
         "highWatermark": 10                 // default: 10
+    },
+    "maxConnections": 1000,                 // default: nil (unlimited)
+    "connectionTimeouts": {
+        "idle": 60,                         // default: nil (no timeout)
+        "readHeader": 30,                   // default: nil (no timeout)
+        "readBody": 60                      // default: nil (no timeout)
     }
 }
 ```
diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift
index 349116c..3a419bf 100644
--- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift
+++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift
@@ -93,6 +93,12 @@ extension NIOHTTPServer {
                             try channel.pipeline.syncOperations.addHandler(
                                 serverQuiescingHelper.makeServerChannelHandler(channel: channel)
                             )
+
+                            if let maxConnections = self.configuration.maxConnections {
+                                try channel.pipeline.syncOperations.addHandler(
+                                    ConnectionLimitHandler(maxConnections: maxConnections)
+                                )
+                            }
                         }
                     }.bind(host: host, port: port) { channel in
                         self.setupHTTP1_1Connection(
@@ -131,6 +137,11 @@ extension NIOHTTPServer {
             try channel.pipeline.syncOperations.addHandler(HTTP1ToHTTPServerCodec(secure: isSecure))
             try channel.pipeline.syncOperations.addHandler(HTTPKeepAliveHandler())
 
+            try channel
+                .pipeline
+                .syncOperations
+                .addTimeoutHandlers(self.configuration.connectionTimeouts)
+
             return try NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>(
                 wrappingChannelSynchronously: channel,
                 configuration: asyncChannelConfiguration
diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift
index 48ff7b0..6583fa7 100644
--- a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift
+++ b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift
@@ -210,6 +210,12 @@ extension NIOHTTPServer {
                             try channel.pipeline.syncOperations.addHandler(
                                 serverQuiescingHelper.makeServerChannelHandler(channel: channel)
                             )
+
+                            if let maxConnections = self.configuration.maxConnections {
+                                try channel.pipeline.syncOperations.addHandler(
+                                    ConnectionLimitHandler(maxConnections: maxConnections)
+                                )
+                            }
                         }
                     }.bind(host: host, port: port) { channel in
                         self.setupSecureUpgradeConnectionChildChannel(
@@ -249,7 +255,7 @@ extension NIOHTTPServer {
             try channel.pipeline.syncOperations.configureAsyncHTTP2Pipeline(
                 mode: .server,
                 connectionManagerConfiguration: .init(
-                    maxIdleTime: nil,
+                    maxIdleTime: self.configuration.connectionTimeouts.idle.map { TimeAmount($0) },
                     maxAge: nil,
                     maxGraceTime: configuration.gracefulShutdown.maximumGracefulShutdownDuration
                         .map { TimeAmount($0) },
@@ -263,6 +269,12 @@ extension NIOHTTPServer {
                                 HTTP2FramePayloadToHTTPServerCodec()
                             )
 
+                        // Add read header and body timeouts per-stream for HTTP/2
+                        try http2StreamChannel
+                            .pipeline
+                            .syncOperations
+                            .addReadTimeoutHandlers(self.configuration.connectionTimeouts)
+
                         return try NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>(
                             wrappingChannelSynchronously: http2StreamChannel,
                             configuration: .init(
diff --git a/Sources/NIOHTTPServer/NIOHTTPServer.swift b/Sources/NIOHTTPServer/NIOHTTPServer.swift
index 3f72b69..890c6d9 100644
--- a/Sources/NIOHTTPServer/NIOHTTPServer.swift
+++ b/Sources/NIOHTTPServer/NIOHTTPServer.swift
@@ -329,6 +329,41 @@ public struct NIOHTTPServer: HTTPServer {
             }
         }
     }
+
+}
+
+@available(anyAppleOS 26.0, *)
+extension ChannelPipeline.SynchronousOperations {
+    /// Adds timeout handlers (idle, read header, read body) to the channel pipeline.
+    ///
+    /// Only handlers for non-nil timeouts are installed. Called for HTTP/1.1 connection channels.
+    func addTimeoutHandlers(_ timeouts: NIOHTTPServerConfiguration.ConnectionTimeouts) throws {
+        try self.addIdleTimeoutHandlers(timeouts)
+        try self.addReadTimeoutHandlers(timeouts)
+    }
+
+    /// Adds the connection idle timeout handler to the channel. Used by HTTP/1.1 connection
+    /// channels. (HTTP/2 delegates idle handling to `NIOHTTP2ServerConnectionManagementHandler`'s
+    /// `maxIdleTime`, which is stream-aware.)
+    func addIdleTimeoutHandlers(_ timeouts: NIOHTTPServerConfiguration.ConnectionTimeouts) throws {
+        if let idle = timeouts.idle {
+            try self.addHandler(
+                ConnectionIdleTimeoutHandler(timeout: TimeAmount(idle))
+            )
+        }
+    }
+
+    /// Adds only read header and body timeout handlers to the channel. Used for HTTP/1.1
+    /// connection channels and HTTP/2 per-stream channels.
+    func addReadTimeoutHandlers(_ timeouts: NIOHTTPServerConfiguration.ConnectionTimeouts) throws {
+        let readHeader = timeouts.readHeader.map { TimeAmount($0) }
+        let readBody = timeouts.readBody.map { TimeAmount($0) }
+        if readHeader != nil || readBody != nil {
+            try self.addHandler(
+                RequestTimeoutHandler(readHeaderTimeout: readHeader, readBodyTimeout: readBody)
+            )
+        }
+    }
 }
 
 @available(anyAppleOS 26.0, *)
diff --git a/Sources/NIOHTTPServer/TimeoutHandlers.swift b/Sources/NIOHTTPServer/TimeoutHandlers.swift
new file mode 100644
index 0000000..8bc9178
--- /dev/null
+++ b/Sources/NIOHTTPServer/TimeoutHandlers.swift
@@ -0,0 +1,143 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift HTTP Server open source project
+//
+// Copyright (c) 2026 Apple Inc. and the Swift HTTP Server project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of Swift HTTP Server project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+import NIOHTTPTypes
+
+/// A channel handler that closes an HTTP/1.1 connection after a period in which no request is in
+/// flight.
+///
+/// The timer runs only between requests: it is scheduled when the channel becomes active and
+/// after each response `.end` is written. It is cancelled when an inbound request `.head` is
+/// observed. While a request is being processed, request-level timeouts (see
+/// ``RequestTimeoutHandler``) are responsible for protecting the server.
+///
+/// This handler is used on the per-connection channel for HTTP/1.1 only. For HTTP/2, idle
+/// behaviour is delegated to `NIOHTTP2ServerConnectionManagementHandler`'s `maxIdleTime`, which
+/// already understands stream lifecycle.
+final class ConnectionIdleTimeoutHandler: ChannelDuplexHandler, RemovableChannelHandler {
+    typealias InboundIn = HTTPRequestPart
+    typealias InboundOut = HTTPRequestPart
+    typealias OutboundIn = HTTPResponsePart
+    typealias OutboundOut = HTTPResponsePart
+
+    private let timeout: TimeAmount
+    private var scheduledTimeout: Scheduled<Void>?
+
+    init(timeout: TimeAmount) {
+        self.timeout = timeout
+    }
+
+    func channelActive(context: ChannelHandlerContext) {
+        // Connection just opened, no request yet — start the idle timer.
+        self.scheduleTimeout(context: context)
+        context.fireChannelActive()
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        let part = self.unwrapInboundIn(data)
+        if case .head = part {
+            // A request just started; pause idle until the response is fully written.
+            self.scheduledTimeout?.cancel()
+            self.scheduledTimeout = nil
+        }
+        context.fireChannelRead(data)
+    }
+
+    func write(context: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
+        let part = self.unwrapOutboundIn(data)
+        context.write(data, promise: promise)
+        if case .end = part {
+            // The response is complete; the connection is now between requests, so re-arm idle.
+            self.scheduleTimeout(context: context)
+        }
+    }
+
+    func handlerRemoved(context: ChannelHandlerContext) {
+        self.scheduledTimeout?.cancel()
+        self.scheduledTimeout = nil
+    }
+
+    private func scheduleTimeout(context: ChannelHandlerContext) {
+        self.scheduledTimeout?.cancel()
+        self.scheduledTimeout = context.eventLoop.assumeIsolated().scheduleTask(in: self.timeout) {
+            context.close(promise: nil)
+        }
+    }
+}
+
+/// A channel handler that enforces timeouts on receiving request headers and body.
+///
+/// State machine:
+/// - On channel active, a header timeout is scheduled (if configured).
+/// - When `.head` is received, the header timeout is cancelled and a body timeout is scheduled
+///   (if configured).
+/// - When `.end` is received, the body timeout is cancelled and the header timeout is rescheduled
+///   so that the next request on a keep-alive connection is also protected. (For HTTP/2 streams
+///   this is a no-op in practice: each stream sees only one request and is closed shortly after
+///   `.end`.)
+///
+/// If either timeout fires, the connection is closed.
+final class RequestTimeoutHandler: ChannelInboundHandler, RemovableChannelHandler {
+    typealias InboundIn = HTTPRequestPart
+
+    private let readHeaderTimeout: TimeAmount?
+    private let readBodyTimeout: TimeAmount?
+    private var scheduledTimeout: Scheduled<Void>?
+
+    init(readHeaderTimeout: TimeAmount?, readBodyTimeout: TimeAmount?) {
+        self.readHeaderTimeout = readHeaderTimeout
+        self.readBodyTimeout = readBodyTimeout
+    }
+
+    func channelActive(context: ChannelHandlerContext) {
+        if let readHeaderTimeout {
+            self.scheduleTimeout(readHeaderTimeout, context: context)
+        }
+        context.fireChannelActive()
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        let part = self.unwrapInboundIn(data)
+        switch part {
+        case .head:
+            self.scheduledTimeout?.cancel()
+            self.scheduledTimeout = nil
+            if let readBodyTimeout {
+                self.scheduleTimeout(readBodyTimeout, context: context)
+            }
+        case .body:
+            break
+        case .end:
+            self.scheduledTimeout?.cancel()
+            self.scheduledTimeout = nil
+            // Re-arm the header timer so the next request on this connection is also protected.
+            if let readHeaderTimeout {
+                self.scheduleTimeout(readHeaderTimeout, context: context)
+            }
+        }
+        context.fireChannelRead(data)
+    }
+
+    func handlerRemoved(context: ChannelHandlerContext) {
+        self.scheduledTimeout?.cancel()
+        self.scheduledTimeout = nil
+    }
+
+    private func scheduleTimeout(_ timeout: TimeAmount, context: ChannelHandlerContext) {
+        self.scheduledTimeout = context.eventLoop.assumeIsolated().scheduleTask(in: timeout) {
+            context.close(promise: nil)
+        }
+    }
+}
diff --git a/Tests/NIOHTTPServerTests/ConnectionBackpressureConfigurationTests.swift b/Tests/NIOHTTPServerTests/ConnectionBackpressureConfigurationTests.swift
new file mode 100644
index 0000000..3904424
--- /dev/null
+++ b/Tests/NIOHTTPServerTests/ConnectionBackpressureConfigurationTests.swift
@@ -0,0 +1,127 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift HTTP Server open source project
+//
+// Copyright (c) 2026 Apple Inc. and the Swift HTTP Server project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of Swift HTTP Server project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import Testing
+
+@testable import NIOHTTPServer
+
+@Suite("Connection Backpressure Configuration")
+struct ConnectionBackpressureConfigurationTests {
+    @available(anyAppleOS 26.0, *)
+    @Test("maxConnections nil is the default")
+    func maxConnectionsNilIsDefault() throws {
+        let config = try NIOHTTPServerConfiguration(
+            bindTarget: .hostAndPort(host: "127.0.0.1", port: 0),
+            supportedHTTPVersions: [.http1_1],
+            transportSecurity: .plaintext
+        )
+        #expect(config.maxConnections == nil)
+    }
+
+    @available(anyAppleOS 26.0, *)
+    @Test("ConnectionTimeouts defaults has expected values")
+    func connectionTimeoutsDefaults() {
+        let timeouts = NIOHTTPServerConfiguration.ConnectionTimeouts.defaults
+        #expect(timeouts.idle == .seconds(60))
+        #expect(timeouts.readHeader == .seconds(30))
+        #expect(timeouts.readBody == .seconds(60))
+    }
+
+    @available(anyAppleOS 26.0, *)
+    @Test("Valid maxConnections is accepted")
+    func validMaxConnectionsAccepted() throws {
+        var config = try NIOHTTPServerConfiguration(
+            bindTarget: .hostAndPort(host: "127.0.0.1", port: 0),
+            supportedHTTPVersions: [.http1_1],
+            transportSecurity: .plaintext
+        )
+        config.maxConnections = 100
+        #expect(config.maxConnections == 100)
+    }
+
+    @available(anyAppleOS 26.0, *)
+    @Test("Custom ConnectionTimeouts are preserved")
+    func customConnectionTimeouts() throws {
+        var config = try NIOHTTPServerConfiguration(
+            bindTarget: .hostAndPort(host: "127.0.0.1", port: 0),
+            supportedHTTPVersions: [.http1_1],
+            transportSecurity: .plaintext
+        )
+        config.connectionTimeouts = .init(idle: .seconds(10), readHeader: .seconds(5), readBody: nil)
+        #expect(config.connectionTimeouts.idle == .seconds(10))
+        #expect(config.connectionTimeouts.readHeader == .seconds(5))
+        #expect(config.connectionTimeouts.readBody == nil)
+    }
+}
+
+#if Configuration
+import Configuration
+
+@Suite("Connection Backpressure SwiftConfiguration")
+struct ConnectionBackpressureSwiftConfigurationTests {
+    @available(anyAppleOS 26.0, *)
+    @Test("SwiftConfiguration parses maxConnections")
+    func parsesMaxConnections() throws {
+        let provider = InMemoryProvider(values: [
+            "bindTarget.host": "127.0.0.1",
+            "bindTarget.port": 8080,
+            "http.versions": .init(.stringArray(["http1_1"]), isSecret: false),
+            "transportSecurity.mode": "plaintext",
+            "maxConnections": 500,
+        ])
+        let config = ConfigReader(provider: provider)
+        let serverConfig = try NIOHTTPServerConfiguration(config: config)
+
+        #expect(serverConfig.maxConnections == 500)
+    }
+
+    @available(anyAppleOS 26.0, *)
+    @Test("SwiftConfiguration parses connectionTimeouts")
+    func parsesConnectionTimeouts() throws {
+        let provider = InMemoryProvider(values: [
+            "bindTarget.host": "127.0.0.1",
+            "bindTarget.port": 8080,
+            "http.versions": .init(.stringArray(["http1_1"]), isSecret: false),
+            "transportSecurity.mode": "plaintext",
+            "connectionTimeouts.idle": 120,
+            "connectionTimeouts.readHeader": 15,
+            "connectionTimeouts.readBody": 45,
+        ])
+        let config = ConfigReader(provider: provider)
+        let serverConfig = try NIOHTTPServerConfiguration(config: config)
+
+        #expect(serverConfig.connectionTimeouts.idle == .seconds(120))
+        #expect(serverConfig.connectionTimeouts.readHeader == .seconds(15))
+        #expect(serverConfig.connectionTimeouts.readBody == .seconds(45))
+    }
+
+    @available(anyAppleOS 26.0, *)
+    @Test("SwiftConfiguration uses defaults for absent fields")
+    func usesDefaultsForAbsentFields() throws {
+        let provider = InMemoryProvider(values: [
+            "bindTarget.host": "127.0.0.1",
+            "bindTarget.port": 8080,
+            "http.versions": .init(.stringArray(["http1_1"]), isSecret: false),
+            "transportSecurity.mode": "plaintext",
+        ])
+        let config = ConfigReader(provider: provider)
+        let serverConfig = try NIOHTTPServerConfiguration(config: config)
+
+        #expect(serverConfig.maxConnections == nil)
+        #expect(serverConfig.connectionTimeouts.idle == nil)
+        #expect(serverConfig.connectionTimeouts.readHeader == nil)
+        #expect(serverConfig.connectionTimeouts.readBody == nil)
+    }
+}
+#endif  // Configuration
diff --git a/Tests/NIOHTTPServerTests/ConnectionBackpressureEndToEndTests.swift b/Tests/NIOHTTPServerTests/ConnectionBackpressureEndToEndTests.swift
new file mode 100644
index 0000000..3e7579c
--- /dev/null
+++ b/Tests/NIOHTTPServerTests/ConnectionBackpressureEndToEndTests.swift
@@ -0,0 +1,203 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift HTTP Server open source project
+//
+// Copyright (c) 2026 Apple Inc. and the Swift HTTP Server project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of Swift HTTP Server project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import HTTPAPIs
+import Logging
+import NIOCore
+import NIOPosix
+import Synchronization
+import Testing
+
+@testable import NIOHTTPServer
+
+@Suite("Connection Backpressure End-to-End")
+struct ConnectionBackpressureEndToEndTests {
+    let serverLogger = Logger(label: "ConnectionBackpressureE2ETests")
+
+    @available(anyAppleOS 26.0, *)
+    @Test("Requests succeed under connection limit")
+    func requestsSucceedUnderConnectionLimit() async throws {
+        var configuration = try NIOHTTPServerConfiguration(
+            bindTarget: .hostAndPort(host: "127.0.0.1", port: 0),
+            supportedHTTPVersions: [.http1_1],
+            transportSecurity: .plaintext
+        )
+        configuration.maxConnections = 2
+        configuration.connectionTimeouts = .init(idle: nil, readHeader: nil, readBody: nil)
+
+        let server = NIOHTTPServer(
+            logger: self.serverLogger,
+            configuration: configuration
+        )
+
+        try await confirmation(expectedCount: 2) { responseReceived in
+            try await NIOHTTPServerTests.withServer(
+                server: server,
+                serverHandler: HTTPServerClosureRequestHandler { _, _, reader, responseSender in
+                    try await NIOHTTPServerTests.echoResponse(
+                        readUpTo: 1024,
+                        reader: reader,
+                        sender: responseSender
+                    )
+                },
+                body: { serverAddress in
+                    try await withThrowingTaskGroup { group in
+                        for _ in 0..<2 {
+                            group.addTask {
+                                let client = try await ClientBootstrap(
+                                    group: .singletonMultiThreadedEventLoopGroup
+                                ).connectToTestHTTP1Server(at: serverAddress)
+
+                                try await client.executeThenClose { inbound, outbound in
+                                    try await outbound.write(
+                                        .head(.init(method: .get, scheme: "http", authority: "", path: "/"))
+                                    )
+                                    try await outbound.write(.end(nil))
+
+                                    try await NIOHTTPServerTests.validateResponse(
+                                        inbound,
+                                        expectedHead: [NIOHTTPServerTests.responseHead(status: .ok, for: .http1_1)],
+                                        expectedBody: [],
+                                        expectStreamEnd: false
+                                    )
+
+                                    responseReceived()
+                                }
+                            }
+                        }
+
+                        try await group.waitForAll()
+                    }
+                }
+            )
+        }
+    }
+
+    @available(anyAppleOS 26.0, *)
+    @Test("More connections than maxConnections all eventually complete")
+    func moreConnectionsThanLimitAllComplete() async throws {
+        var configuration = try NIOHTTPServerConfiguration(
+            bindTarget: .hostAndPort(host: "127.0.0.1", port: 0),
+            supportedHTTPVersions: [.http1_1],
+            transportSecurity: .plaintext
+        )
+        configuration.maxConnections = 2
+        configuration.connectionTimeouts = .init(idle: nil, readHeader: nil, readBody: nil)
+
+        let server = NIOHTTPServer(
+            logger: self.serverLogger,
+            configuration: configuration
+        )
+
+        // Open 5 connections with maxConnections: 2. All should eventually complete
+        // as the connection limit handler releases slots when connections close.
+        let numConnections = 5
+        try await confirmation(expectedCount: numConnections) { responseReceived in
+            try await NIOHTTPServerTests.withServer(
+                server: server,
+                serverHandler: HTTPServerClosureRequestHandler { _, _, reader, responseSender in
+                    try await NIOHTTPServerTests.echoResponse(
+                        readUpTo: 1024,
+                        reader: reader,
+                        sender: responseSender
+                    )
+                },
+                body: { serverAddress in
+                    await withThrowingTaskGroup { group in
+                        for _ in 0..<numConnections {
+                            group.addTask {
+                                let client = try await ClientBootstrap(
+                                    group: .singletonMultiThreadedEventLoopGroup
+                                ).connectToTestHTTP1Server(at: serverAddress)
+
+                                try await client.executeThenClose { inbound, outbound in
+                                    try await outbound.write(
+                                        .head(.init(method: .get, scheme: "http", authority: "", path: "/"))
+                                    )
+                                    try await outbound.write(.end(nil))
+
+                                    try await NIOHTTPServerTests.validateResponse(
+                                        inbound,
+                                        expectedHead: [NIOHTTPServerTests.responseHead(status: .ok, for: .http1_1)],
+                                        expectedBody: [],
+                                        expectStreamEnd: false
+                                    )
+
+                                    responseReceived()
+                                }
+                            }
+                        }
+                    }
+                }
+            )
+        }
+    }
+
+    @available(anyAppleOS 26.0, *)
+    @Test("No connection limit by default")
+    func noConnectionLimitByDefault() async throws {
+        var configuration = try NIOHTTPServerConfiguration(
+            bindTarget: .hostAndPort(host: "127.0.0.1", port: 0),
+            supportedHTTPVersions: [.http1_1],
+            transportSecurity: .plaintext
+        )
+        configuration.connectionTimeouts = .init(idle: nil, readHeader: nil, readBody: nil)
+
+        let server = NIOHTTPServer(
+            logger: self.serverLogger,
+            configuration: configuration
+        )
+
+        let numConnections = 5
+        try await confirmation(expectedCount: numConnections) { responseReceived in
+            try await NIOHTTPServerTests.withServer(
+                server: server,
+                serverHandler: HTTPServerClosureRequestHandler { _, _, reader, responseSender in
+                    try await NIOHTTPServerTests.echoResponse(
+                        readUpTo: 1024,
+                        reader: reader,
+                        sender: responseSender
+                    )
+                },
+                body: { serverAddress in
+                    await withThrowingTaskGroup { group in
+                        for _ in 0..<numConnections {
+                            group.addTask {
+                                let client = try await ClientBootstrap(
+                                    group: .singletonMultiThreadedEventLoopGroup
+                                ).connectToTestHTTP1Server(at: serverAddress)
+
+                                try await client.executeThenClose { inbound, outbound in
+                                    try await outbound.write(
+                                        .head(.init(method: .get, scheme: "http", authority: "", path: "/"))
+                                    )
+                                    try await outbound.write(.end(nil))
+
+                                    try await NIOHTTPServerTests.validateResponse(
+                                        inbound,
+                                        expectedHead: [NIOHTTPServerTests.responseHead(status: .ok, for: .http1_1)],
+                                        expectedBody: [],
+                                        expectStreamEnd: false
+                                    )
+
+                                    responseReceived()
+                                }
+                            }
+                        }
+                    }
+                }
+            )
+        }
+    }
+}
diff --git a/Tests/NIOHTTPServerTests/ConnectionLimitHandlerTests.swift b/Tests/NIOHTTPServerTests/ConnectionLimitHandlerTests.swift
new file mode 100644
index 0000000..62e6dff
--- /dev/null
+++ b/Tests/NIOHTTPServerTests/ConnectionLimitHandlerTests.swift
@@ -0,0 +1,158 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift HTTP Server open source project
+//
+// Copyright (c) 2026 Apple Inc. and the Swift HTTP Server project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of Swift HTTP Server project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+import NIOEmbedded
+import NIOHTTPTypes
+import Testing
+
+@testable import NIOHTTPServer
+
+@Suite("ConnectionLimitHandler")
+struct ConnectionLimitHandlerTests {
+
+    @Test("Connections under limit are accepted")
+    func connectionsUnderLimitAreAccepted() async throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionLimitHandler(maxConnections: 3)
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        // Create and write 3 child channels
+        var children: [EmbeddedChannel] = []
+        for _ in 0..<3 {
+            let child = EmbeddedChannel()
+            children.append(child)
+            try channel.writeInbound(child as Channel)
+        }
+
+        // All 3 should have been forwarded
+        for _ in 0..<3 {
+            let forwarded = try channel.readInbound(as: Channel.self)
+            #expect(forwarded != nil)
+        }
+    }
+
+    @Test("Connections at limit block reads")
+    func connectionsAtLimitBlockReads() async throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionLimitHandler(maxConnections: 2)
+
+        // Add a handler closer to the head that counts read() calls that pass through.
+        let readCounter = ReadCountingHandler()
+        try channel.pipeline.syncOperations.addHandler(readCounter)
+        try channel.pipeline.syncOperations.addHandler(handler, position: .last)
+
+        // Open 2 connections to fill the limit
+        let child1 = EmbeddedChannel()
+        let child2 = EmbeddedChannel()
+        try channel.writeInbound(child1 as Channel)
+        try channel.writeInbound(child2 as Channel)
+
+        // Trigger a read while within the acceptable number of connections: it should be forwarded.
+        channel.read()
+        channel.embeddedEventLoop.run()
+
+        #expect(readCounter.readCount == 1, "Read should be forwarded when under the limit")
+
+        // Open a third connection - this will be above the limit, so stop forwarding reads.
+        let child3 = EmbeddedChannel()
+        try channel.writeInbound(child3 as Channel)
+
+        // Now at capacity — a third read should be blocked
+        channel.pipeline.read()
+        #expect(readCounter.readCount == 1, "Third read should NOT be forwarded when at capacity")
+    }
+
+    @Test("Connections resume after close")
+    func connectionsResumeAfterClose() async throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionLimitHandler(maxConnections: 1)
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        // Open 1 connection (at limit)
+        let child1 = EmbeddedChannel()
+        try channel.writeInbound(child1 as Channel)
+        _ = try channel.readInbound(as: Channel.self)
+
+        // Close the child connection
+        try await child1.close().get()
+
+        // Run pending tasks on the event loop
+        channel.embeddedEventLoop.run()
+
+        // Now we should be able to accept a new connection
+        let child2 = EmbeddedChannel()
+        try channel.writeInbound(child2 as Channel)
+        let forwarded = try channel.readInbound(as: Channel.self)
+        #expect(forwarded != nil)
+    }
+
+    @Test("Multiple connections close and resume")
+    func multipleConnectionsCloseAndResume() async throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionLimitHandler(maxConnections: 3)
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        // Open 3 connections
+        var children: [EmbeddedChannel] = []
+        for _ in 0..<3 {
+            let child = EmbeddedChannel()
+            children.append(child)
+            try channel.writeInbound(child as Channel)
+            _ = try channel.readInbound(as: Channel.self)
+        }
+
+        // Close 2 of them
+        try await children[0].close().get()
+        try await children[1].close().get()
+        channel.embeddedEventLoop.run()
+
+        // Should be able to accept 2 more
+        for _ in 0..<2 {
+            let child = EmbeddedChannel()
+            try channel.writeInbound(child as Channel)
+            let forwarded = try channel.readInbound(as: Channel.self)
+            #expect(forwarded != nil)
+        }
+    }
+
+    @Test("Handler does not interfere when under limit")
+    func handlerDoesNotInterfereUnderLimit() async throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionLimitHandler(maxConnections: 100)
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        // Open 5 connections — well under the limit
+        for _ in 0..<5 {
+            let child = EmbeddedChannel()
+            try channel.writeInbound(child as Channel)
+            let forwarded = try channel.readInbound(as: Channel.self)
+            #expect(forwarded != nil)
+        }
+    }
+}
+
+/// A handler that counts how many `read()` calls pass through it.
+/// Placed before (closer to the head) the `ConnectionLimitHandler` in the pipeline
+/// so it observes reads that the limiter forwards toward the socket.
+private final class ReadCountingHandler: ChannelOutboundHandler {
+    typealias OutboundIn = Any
+
+    var readCount: Int = 0
+
+    func read(context: ChannelHandlerContext) {
+        self.readCount += 1
+        context.read()
+    }
+}
diff --git a/Tests/NIOHTTPServerTests/TimeoutHandlerTests.swift b/Tests/NIOHTTPServerTests/TimeoutHandlerTests.swift
new file mode 100644
index 0000000..b521bf3
--- /dev/null
+++ b/Tests/NIOHTTPServerTests/TimeoutHandlerTests.swift
@@ -0,0 +1,342 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift HTTP Server open source project
+//
+// Copyright (c) 2026 Apple Inc. and the Swift HTTP Server project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of Swift HTTP Server project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import HTTPTypes
+import NIOCore
+import NIOEmbedded
+import NIOHTTPTypes
+import Testing
+
+@testable import NIOHTTPServer
+
+@Suite("ConnectionIdleTimeoutHandler")
+struct ConnectionIdleTimeoutHandlerTests {
+
+    @Test("Connection closed after idle timeout with no request")
+    func closedAfterIdleTimeout() throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionIdleTimeoutHandler(timeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        // Advance past the timeout with no request in flight
+        channel.embeddedEventLoop.advanceTime(by: .seconds(6))
+
+        #expect(!channel.isActive)
+    }
+
+    @Test("Idle timer is cancelled while a request is in flight")
+    func idleCancelledWhileRequestInFlight() throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionIdleTimeoutHandler(timeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        // Send a request head before the idle timeout would fire.
+        let head = HTTPRequest(method: .post, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        // Advance well past the original idle window. Idle should not fire because a request is
+        // in flight (response not yet written).
+        channel.embeddedEventLoop.advanceTime(by: .seconds(60))
+
+        #expect(channel.isActive)
+    }
+
+    @Test("Body parts do not reset idle (because idle is paused)")
+    func bodyPartsDoNotMatter() throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionIdleTimeoutHandler(timeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        let head = HTTPRequest(method: .post, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+        try channel.writeInbound(HTTPRequestPart.body(ByteBuffer(bytes: [1, 2, 3])))
+
+        // Idle is paused while a request is in flight; it doesn't matter that we got body bytes.
+        channel.embeddedEventLoop.advanceTime(by: .seconds(10))
+        #expect(channel.isActive)
+    }
+
+    @Test("Idle timer is rearmed after response end (between requests)")
+    func idleRearmedAfterResponseEnd() throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionIdleTimeoutHandler(timeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        // Process a request fully.
+        let head = HTTPRequest(method: .get, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+        try channel.writeInbound(HTTPRequestPart.end(nil))
+        try channel.writeOutbound(HTTPResponsePart.head(HTTPResponse(status: .ok)))
+        try channel.writeOutbound(HTTPResponsePart.end(nil))
+
+        // No new request — advance past the idle window. Connection should close.
+        channel.embeddedEventLoop.advanceTime(by: .seconds(6))
+        #expect(!channel.isActive)
+    }
+
+    @Test("Idle timer is cancelled when next request begins on a keep-alive connection")
+    func idleCancelledOnNextRequest() throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionIdleTimeoutHandler(timeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        // First request/response cycle.
+        let head = HTTPRequest(method: .get, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+        try channel.writeInbound(HTTPRequestPart.end(nil))
+        try channel.writeOutbound(HTTPResponsePart.head(HTTPResponse(status: .ok)))
+        try channel.writeOutbound(HTTPResponsePart.end(nil))
+
+        // Wait partway, then start a second request before idle fires.
+        channel.embeddedEventLoop.advanceTime(by: .seconds(4))
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        // Advance well past — idle should be paused again.
+        channel.embeddedEventLoop.advanceTime(by: .seconds(60))
+        #expect(channel.isActive)
+    }
+
+    @Test("Cleanup on handler removal")
+    func cleanupOnHandlerRemoval() throws {
+        let channel = EmbeddedChannel()
+        let handler = ConnectionIdleTimeoutHandler(timeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        _ = channel.pipeline.syncOperations.removeHandler(handler)
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(10))
+
+        #expect(channel.isActive)
+    }
+}
+
+@Suite("RequestTimeoutHandler")
+struct RequestTimeoutHandlerTests {
+
+    // MARK: - Header timeout tests
+
+    @Test("Headers received within timeout — connection stays open")
+    func headersReceivedWithinTimeout() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: .seconds(5), readBodyTimeout: nil)
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        let head = HTTPRequest(method: .get, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(10))
+
+        #expect(channel.isActive)
+    }
+
+    @Test("Headers not received within timeout — connection closed")
+    func headersNotReceivedWithinTimeout() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: .seconds(5), readBodyTimeout: nil)
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(6))
+
+        #expect(!channel.isActive)
+    }
+
+    // MARK: - Body timeout tests
+
+    @Test("Body completed within timeout — connection stays open")
+    func bodyCompletedWithinTimeout() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: nil, readBodyTimeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        let head = HTTPRequest(method: .post, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        try channel.writeInbound(HTTPRequestPart.end(nil))
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(10))
+
+        #expect(channel.isActive)
+    }
+
+    @Test("Body not completed within timeout — connection closed")
+    func bodyNotCompletedWithinTimeout() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: nil, readBodyTimeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        let head = HTTPRequest(method: .post, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(6))
+
+        #expect(!channel.isActive)
+    }
+
+    @Test("Body parts do not reset timeout")
+    func bodyPartsDoNotResetTimeout() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: nil, readBodyTimeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        let head = HTTPRequest(method: .post, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(2))
+        try channel.writeInbound(HTTPRequestPart.body(ByteBuffer(bytes: [1, 2, 3])))
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(2))
+        try channel.writeInbound(HTTPRequestPart.body(ByteBuffer(bytes: [4, 5, 6])))
+
+        // Total 6s since head — past the 5s timeout
+        channel.embeddedEventLoop.advanceTime(by: .seconds(2))
+
+        #expect(!channel.isActive)
+    }
+
+    // MARK: - Combined timeout tests
+
+    @Test("Both timeouts configured — header then body")
+    func bothTimeoutsHeaderThenBody() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: .seconds(5), readBodyTimeout: .seconds(10))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        // Send head within header timeout
+        channel.embeddedEventLoop.advanceTime(by: .seconds(3))
+        let head = HTTPRequest(method: .post, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        // Send end within body timeout
+        channel.embeddedEventLoop.advanceTime(by: .seconds(8))
+        try channel.writeInbound(HTTPRequestPart.end(nil))
+
+        // Right after `.end`, the header timeout is re-armed for the next request. Send a fresh
+        // head before that timer fires so the connection stays open.
+        channel.embeddedEventLoop.advanceTime(by: .seconds(3))
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        // Body timer is now ticking on the second request — finish within the body timeout.
+        channel.embeddedEventLoop.advanceTime(by: .seconds(5))
+        try channel.writeInbound(HTTPRequestPart.end(nil))
+
+        #expect(channel.isActive)
+    }
+
+    @Test("Both timeouts configured — header timeout fires")
+    func bothTimeoutsHeaderFires() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: .seconds(5), readBodyTimeout: .seconds(10))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(6))
+
+        #expect(!channel.isActive)
+    }
+
+    @Test("Both timeouts configured — body timeout fires")
+    func bothTimeoutsBodyFires() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: .seconds(5), readBodyTimeout: .seconds(10))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        let head = HTTPRequest(method: .post, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(11))
+
+        #expect(!channel.isActive)
+    }
+
+    @Test("Header timeout is re-armed after end so subsequent requests are protected")
+    func headerTimeoutRearmedAfterEnd() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: .seconds(5), readBodyTimeout: nil)
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        // First request completes successfully within the header timeout window.
+        let head = HTTPRequest(method: .get, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+        try channel.writeInbound(HTTPRequestPart.end(nil))
+
+        // Now no second request arrives within the header timeout — connection should be closed.
+        channel.embeddedEventLoop.advanceTime(by: .seconds(6))
+
+        #expect(!channel.isActive)
+    }
+
+    // MARK: - Cleanup
+
+    @Test("Cleanup on handler removal during header phase")
+    func cleanupOnHandlerRemovalDuringHeaderPhase() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: .seconds(5), readBodyTimeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        _ = channel.pipeline.syncOperations.removeHandler(handler)
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(10))
+
+        #expect(channel.isActive)
+    }
+
+    @Test("Cleanup on handler removal during body phase")
+    func cleanupOnHandlerRemovalDuringBodyPhase() throws {
+        let channel = EmbeddedChannel()
+        let handler = RequestTimeoutHandler(readHeaderTimeout: .seconds(5), readBodyTimeout: .seconds(5))
+        try channel.pipeline.syncOperations.addHandler(handler)
+
+        try channel.connect(to: .init(ipAddress: "127.0.0.1", port: 8080)).wait()
+
+        let head = HTTPRequest(method: .post, scheme: "http", authority: "", path: "/")
+        try channel.writeInbound(HTTPRequestPart.head(head))
+
+        _ = channel.pipeline.syncOperations.removeHandler(handler)
+
+        channel.embeddedEventLoop.advanceTime(by: .seconds(10))
+
+        #expect(channel.isActive)
+    }
+}