diff --git a/Sources/NIOHTTPServer/HTTPKeepAliveHandler.swift b/Sources/NIOHTTPServer/HTTPKeepAliveHandler.swift index 26f7cb6..d8640fa 100644 --- a/Sources/NIOHTTPServer/HTTPKeepAliveHandler.swift +++ b/Sources/NIOHTTPServer/HTTPKeepAliveHandler.swift @@ -12,11 +12,15 @@ // //===----------------------------------------------------------------------===// +import NIOConcurrencyHelpers import NIOCore import NIOHTTPTypes -/// A NIO channel handler that ensures HTTP/1.1 keep-alive semantics are honored when -/// the server starts writing a response before the request body has been fully read. +/// A NIO channel handler that ensures HTTP/1.1 keep-alive semantics are honored +/// when the server starts writing a response before the request body has been +/// fully read, and that reacts to the connection-scoped close flag by amending +/// the next response head with `Connection: close` and closing the channel after +/// the response is written. /// /// The handler buffers final response parts (head + any body fragments + end) when /// they are written before the request `.end` has been received. The buffer is @@ -26,13 +30,15 @@ import NIOHTTPTypes /// - **`flush`**: an upstream writer (e.g. `NIOAsyncChannelOutboundWriter`) forced a /// flush. /// -/// At each deadline, if request `.end` has arrived, the buffer is flushed as-is and -/// the connection is reusable. If request `.end` has *not* arrived, the head is -/// amended with `Connection: close`, the buffer is flushed, and the connection is -/// closed once response `.end` is written. This protects against clients that keep -/// uploading request body bytes after the response has completed (which would -/// otherwise force the server to drain unbounded data) and gives the client an -/// explicit signal not to pipeline another request on the connection. +/// At each deadline, if request `.end` has arrived and no close has been +/// signalled, the buffer is flushed as-is and the connection is reusable. If +/// request `.end` has *not* arrived, or the shared close flag has been set, the +/// head is amended with `Connection: close`, the buffer is flushed, and the +/// connection is closed once response `.end` is written. This protects against +/// clients that keep uploading request body bytes after the response has +/// completed (which would otherwise force the server to drain unbounded data) +/// and gives the client an explicit signal not to pipeline another request on +/// the connection. /// /// Informational (1xx) responses pass through unchanged and do not affect buffering /// state. @@ -64,32 +70,59 @@ final class HTTPKeepAliveHandler: ChannelDuplexHandler { case streaming } - /// `true` when the request `.end` has been received on the inbound side, or no - /// request is currently in flight. `false` between receiving a request `.head` - /// and its `.end`. - private var requestEndReceived: Bool = true - - /// `true` if we've committed to closing the connection after this response's - /// `.end` is written. Set when the buffer is flushed while request `.end` has - /// not yet arrived (so we add `Connection: close`). Cleared when a new request - /// begins. + /// `true` between receiving a request `.head` and its `.end` — i.e. while the + /// client may still be streaming request body bytes. `false` otherwise: on a + /// fresh handler with no request yet, between requests, and after a request's + /// `.end` has arrived. The final-response decision hinges on this: writing a + /// final response head while a body is in flight risks HTTP/1.1 framing + /// desync on keep-alive, so those writes must be buffered until the flush + /// deadline (`channelReadComplete` / `flush`) and, if the body still hasn't + /// completed by then, sent with `Connection: close`. + private var requestBodyInFlight: Bool = false + + /// `true` if the response head that has been (or will be) written for the + /// current request carries `Connection: close` — either because the close + /// flag was set before the head reached the wire, or because the head was + /// flushed while a request body was still in flight. When true, response + /// `.end` triggers a channel close. + /// + /// If `closeSignalled` becomes true only *after* the head has been streamed + /// (so the header could no longer be amended), this flag stays `false` — + /// the close is still honoured, but via a direct `closeSignalled` check on + /// response `.end` rather than through this flag. + /// + /// Cleared when a new request begins. private var closeAfterResponseEnd: Bool = false private var finalResponseState: FinalResponseState = .notStarted + /// Shared close flag. ``NIOHTTPServer/ConnectionContext/signalConnectionClose()`` + /// sets this synchronously; the handler observes it when writing the next + /// response head and amends with `Connection: close`, then closes the channel + /// after the response `.end`. + let closeFlag: NIOLockedValueBox + + init(closeFlag: NIOLockedValueBox) { + self.closeFlag = closeFlag + } + + private var closeSignalled: Bool { + self.closeFlag.withLockedValue { $0 } + } + func channelRead(context: ChannelHandlerContext, data: NIOAny) { let part = self.unwrapInboundIn(data) switch part { case .head: // Begin a new request. (Any previous request's response must have // completed already since HTTPServerPipelineHandler enforces ordering.) - self.requestEndReceived = false + self.requestBodyInFlight = true self.closeAfterResponseEnd = false self.finalResponseState = .notStarted case .body: break case .end: - self.requestEndReceived = true + self.requestBodyInFlight = false } context.fireChannelRead(data) } @@ -100,8 +133,8 @@ final class HTTPKeepAliveHandler: ChannelDuplexHandler { // `Connection: close`. If request `.end` arrived during the cycle the head // is flushed unchanged; otherwise we amend the head and close after // response `.end`. - if case .buffering = self.finalResponseState { - self.flushBuffer(context: context) + if case let .buffering(head, additional) = self.finalResponseState { + self.flushBuffer(context: context, head: head, additional: additional) } context.fireChannelReadComplete() } @@ -117,10 +150,21 @@ final class HTTPKeepAliveHandler: ChannelDuplexHandler { context.write(data, promise: promise) return } - if self.requestEndReceived { - // Request fully read; stream the response directly. + if !self.requestBodyInFlight { + // No request body in flight; stream the response directly. If a + // close signal has been observed, amend the head with + // `Connection: close` and arrange to close once response `.end` + // is written. + let outboundPart: HTTPResponsePart + if self.closeSignalled, case .head(var response) = part { + response.headerFields[.connection] = "close" + outboundPart = .head(response) + self.closeAfterResponseEnd = true + } else { + outboundPart = part + } self.finalResponseState = .streaming - context.write(data, promise: promise) + context.write(self.wrapOutboundOut(outboundPart), promise: promise) } else { // Start buffering with the head. Additional parts (body, end) the // handler may write before the next deadline are appended below. @@ -134,9 +178,13 @@ final class HTTPKeepAliveHandler: ChannelDuplexHandler { self.finalResponseState = .buffering(head: head, additional: additional) case .streaming: context.write(data, promise: promise) - if case .end = part, self.closeAfterResponseEnd { - // The head we flushed earlier carried `Connection: close`; close - // the connection now that the response is complete. + if case .end = part, self.closeAfterResponseEnd || self.closeSignalled { + // Either the head we flushed earlier carried `Connection: close` + // (`closeAfterResponseEnd`), or a close signal arrived after the + // head was on the wire (`closeSignalled`). In the latter case + // the header can't be added retroactively, but we still honour + // the caller's contract and close the connection now that the + // response is complete. context.flush() context.close(mode: .output, promise: nil) } @@ -145,21 +193,31 @@ final class HTTPKeepAliveHandler: ChannelDuplexHandler { func flush(context: ChannelHandlerContext) { // An upstream writer forced a flush. Same deadline as `channelReadComplete`: - // release any buffered parts, with `Connection: close` if request `.end` - // hasn't arrived. - if case .buffering = self.finalResponseState { - self.flushBuffer(context: context) + // release any buffered parts, with `Connection: close` if the request body + // is still in flight or close was signalled. `flushBuffer` flushes itself, + // so we only propagate the incoming flush when there's no buffer to + // release — otherwise we'd double-flush. + if case let .buffering(head, additional) = self.finalResponseState { + self.flushBuffer(context: context, head: head, additional: additional) + } else { + context.flush() } - context.flush() } - /// Releases buffered response parts to the pipeline. If request `.end` has not - /// yet arrived, amend the head with `Connection: close` and arrange to close - /// the connection once response `.end` is written. - private func flushBuffer(context: ChannelHandlerContext) { - guard case .buffering(var head, let additional) = self.finalResponseState else { return } - - if !self.requestEndReceived { + /// Releases the given buffered response parts to the pipeline. If a request + /// body is still in flight, or if the shared close flag is set, amends the + /// head with `Connection: close` and arranges to close the connection once + /// response `.end` is written. Called only by the two flush deadlines + /// (`channelReadComplete` and `flush`), each of which supplies the buffer + /// contents extracted from the `.buffering` state. + private func flushBuffer( + context: ChannelHandlerContext, + head: BufferedWrite, + additional: [BufferedWrite] + ) { + var head = head + + if self.requestBodyInFlight || self.closeSignalled { // Amend the head with `Connection: close` before flushing. if case .head(var response) = head.part { response.headerFields[.connection] = "close" @@ -180,10 +238,12 @@ final class HTTPKeepAliveHandler: ChannelDuplexHandler { } context.flush() - if sawEnd && self.closeAfterResponseEnd { + if sawEnd && (self.closeAfterResponseEnd || self.closeSignalled) { // The response was fully buffered (head + ... + end) and we have to - // close. Close now (the flush above ensured the writes reached the - // wire). + // close — either we amended the head with `Connection: close` at the + // top of this method (`closeAfterResponseEnd`), or a close signal + // arrived between the head decision and here (`closeSignalled`). + // Close now (the flush above ensured the writes reached the wire). context.close(mode: .output, promise: nil) } } diff --git a/Sources/NIOHTTPServer/HTTPServerCapability+ConnectionCapabilities.swift b/Sources/NIOHTTPServer/HTTPServerCapability+ConnectionCapabilities.swift index ca07244..1515b69 100644 --- a/Sources/NIOHTTPServer/HTTPServerCapability+ConnectionCapabilities.swift +++ b/Sources/NIOHTTPServer/HTTPServerCapability+ConnectionCapabilities.swift @@ -41,4 +41,21 @@ extension HTTPServerCapability { /// The peer's validated certificate chain, when available. var peerCertificateChain: X509.ValidatedCertificateChain? { get async throws } } + + /// A request-context capability that lets a request handler signal that the + /// connection should close after the current response. + /// + /// Servers whose request context conforms to this capability allow handlers + /// to indicate that the underlying connection should be closed once the + /// in-flight response has been sent. Implementations make the signal + /// effective on a best-effort basis (HTTP/1.1 typically appends + /// `Connection: close` and closes the channel; HTTP/2 typically sends + /// `GOAWAY` and lets in-flight streams complete normally). + public protocol CloseableConnection: RequestContext { + /// Signal that the connection should close after the current response. + /// + /// Non-blocking and idempotent. Subsequent calls have no additional + /// effect. + func signalConnectionClose() + } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+Connection.swift b/Sources/NIOHTTPServer/NIOHTTPServer+Connection.swift index 09964da..53fce24 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+Connection.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+Connection.swift @@ -23,7 +23,8 @@ extension NIOHTTPServer { /// A `Connection` is owned by exactly one ``NIOHTTPServerConnectionHandler/handleConnection(connection:context:)`` /// invocation. The handler typically passes it to ``handleRequests(handler:)``, /// which runs the request loop until the peer closes the connection, the - /// server shuts down, or an error occurs. + /// server shuts down, the handler signals close via + /// ``ConnectionContext/signalConnectionClose()``, or an error occurs. /// /// A handler that decides to terminate before any request can simply /// return without calling ``handleRequests(handler:)``: the `Connection` @@ -63,6 +64,7 @@ extension NIOHTTPServer { /// /// Each request received on this connection is dispatched to `handler`. The /// loop returns when the peer closes the connection, the server shuts down, + /// the handler signals close via ``ConnectionContext/signalConnectionClose()``, /// or an error occurs. public consuming func handleRequests( handler: Handler diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift b/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift index d3be12a..2462b68 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift @@ -12,6 +12,7 @@ // //===----------------------------------------------------------------------===// +import NIOConcurrencyHelpers import NIOCore import NIOSSL public import X509 @@ -30,12 +31,14 @@ extension NIOHTTPServer { /// Connection-scoped state. /// /// Carries connection-scoped data such as the negotiated HTTP version, the - /// peer / local addresses, and the peer's validated certificate chain (when - /// applicable). + /// peer / local addresses, the peer's validated certificate chain (when + /// applicable), and the close-signal trigger that + /// ``signalConnectionClose()`` invokes. /// /// User code accesses this state via the corresponding ``RequestContext`` /// capabilities (``HTTPServerCapability/ConnectionInfo``, - /// ``HTTPServerCapability/PeerCertificate``) when handling individual + /// ``HTTPServerCapability/PeerCertificate``, + /// ``HTTPServerCapability/CloseableConnection``) when handling individual /// requests, and directly when implementing an /// ``NIOHTTPServerConnectionHandler``. public struct ConnectionContext: Sendable { @@ -50,16 +53,31 @@ extension NIOHTTPServer { var peerCertificateChainFuture: EventLoopFuture? + /// Per-protocol close mechanism. HTTP/1.1 uses a shared atomic flag the + /// channel's ``HTTPKeepAliveHandler`` reads when writing the next + /// response head. HTTP/2 fires `ChannelShouldQuiesceEvent` on the + /// connection channel; NIO's `NIOHTTP2ServerConnectionManagementHandler` + /// reacts by sending `GOAWAY`, letting in-flight streams complete + /// normally, and finally closing the connection. + let closeBacking: CloseBacking + + enum CloseBacking: Sendable { + case http1_1(closeFlag: NIOLockedValueBox) + case http2(connectionChannel: any Channel) + } + init( httpVersion: HTTPVersion, remoteAddress: NIOHTTPServer.SocketAddress? = nil, localAddress: NIOHTTPServer.SocketAddress? = nil, - peerCertificateChainFuture: EventLoopFuture? = nil + peerCertificateChainFuture: EventLoopFuture? = nil, + closeBacking: CloseBacking ) { self.httpVersion = httpVersion self.remoteAddress = remoteAddress self.localAddress = localAddress self.peerCertificateChainFuture = peerCertificateChainFuture + self.closeBacking = closeBacking } /// The peer's validated certificate chain. Returns `nil` if a custom @@ -74,5 +92,22 @@ extension NIOHTTPServer { return nil } } + + /// Signal that the connection should close after the current response. + /// + /// - On HTTP/1.1, the response carries `Connection: close` and the channel is closed once the response has + /// been written. Note that amending the response with the `close` header will only happen if this function is called + /// before the response `head` has been flushed to the wire. Either way the connection will be closed after the current + /// response, but it may not carry the `Connection: close` header. + /// - On HTTP/2, the connection sends `GOAWAY`; in-flight streams + /// complete normally before the connection closes. + public func signalConnectionClose() { + switch self.closeBacking { + case .http1_1(let closeFlag): + closeFlag.withLockedValue { $0 = true } + case .http2(let connectionChannel): + connectionChannel.pipeline.fireUserInboundEventTriggered(ChannelShouldQuiesceEvent()) + } + } } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift index 494f140..af9e68d 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift @@ -13,6 +13,7 @@ //===----------------------------------------------------------------------===// import Logging +import NIOConcurrencyHelpers import NIOCore import NIOExtras import NIOHTTP1 @@ -23,6 +24,14 @@ import NIOSSL @available(anyAppleOS 26.0, *) extension NIOHTTPServer { + /// An HTTP/1.1 connection vended by the accept loop: the async channel and + /// the close flag the channel's ``HTTPKeepAliveHandler`` shares with the + /// connection context. + struct HTTP1ChildConnection: Sendable { + let asyncChannel: NIOAsyncChannel + let closeFlag: NIOLockedValueBox + } + /// Serves incoming plaintext HTTP/1.1 connections. /// /// Each connection is handled concurrently in its own child task. Individual connection errors are handled within @@ -34,7 +43,7 @@ extension NIOHTTPServer { /// /// - Throws: If an error occurs while iterating the incoming connection stream. func serveInsecureHTTP1_1( - serverChannel: NIOAsyncChannel, Never>, + serverChannel: NIOAsyncChannel, connectionHandler: Handler ) async throws { try await serverChannel.executeThenClose { inbound in @@ -44,10 +53,10 @@ extension NIOHTTPServer { // `inbound`) must be caught and handled directly. let inboundConnectionIterationError = await withDiscardingTaskGroup { group -> (any Error)? in do { - for try await requestChannel in inbound { + for try await child in inbound { group.addTask { await self.dispatchPlaintextHTTP1_1Connection( - requestChannel: requestChannel, + child: child, connectionHandler: connectionHandler ) } @@ -74,13 +83,14 @@ extension NIOHTTPServer { /// `NIOAsyncWriter` is finished cleanly whether or not the connection /// handler called ``Connection/handleRequests(handler:)``. private func dispatchPlaintextHTTP1_1Connection( - requestChannel: sending NIOAsyncChannel, + child: sending HTTP1ChildConnection, connectionHandler: Handler ) async { do { - try await requestChannel.executeThenClose { inbound, outbound in + try await child.asyncChannel.executeThenClose { inbound, outbound in let context = NIOHTTPServer.makeHTTP1ConnectionContext( - requestChannel: requestChannel, + requestChannel: child.asyncChannel, + closeFlag: child.closeFlag, peerCertificateChainFuture: nil ) let connection = Connection( @@ -108,13 +118,13 @@ extension NIOHTTPServer { func setupHTTP1_1ServerChannels( bindTargets: [NIOHTTPServerConfiguration.BindTarget] ) async throws -> [( - NIOAsyncChannel, Never>, ServerQuiescingHelper + NIOAsyncChannel, ServerQuiescingHelper )] { let bootstrap = ServerBootstrap(group: self.eventLoopGroup) .serverChannelOption(.socketOption(.so_reuseaddr), value: 1) var serverChannels = [ - (NIOAsyncChannel, Never>, ServerQuiescingHelper) + (NIOAsyncChannel, ServerQuiescingHelper) ]() do { @@ -163,37 +173,43 @@ extension NIOHTTPServer { return serverChannels } - /// Configures the HTTP/1.1 server pipeline and the keep-alive handler. + /// Configures the HTTP/1.1 server pipeline and the keep-alive handler, sharing + /// a fresh close flag between the keep-alive handler (which observes it when + /// writing the next response head) and the eventually-vended ``HTTP1ChildConnection``. func setupHTTP1_1Connection( channel: any Channel, asyncChannelConfiguration: NIOAsyncChannel.Configuration, isSecure: Bool - ) -> EventLoopFuture> { + ) -> EventLoopFuture { channel.pipeline.configureHTTPServerPipeline().flatMapThrowing { + let closeFlag = NIOLockedValueBox(false) try channel.pipeline.syncOperations.addHandler(HTTP1ToHTTPServerCodec(secure: isSecure)) - try channel.pipeline.syncOperations.addHandler(HTTPKeepAliveHandler()) + try channel.pipeline.syncOperations.addHandler(HTTPKeepAliveHandler(closeFlag: closeFlag)) try channel .pipeline .syncOperations .addTimeoutHandlers(self.configuration.connectionTimeouts) - return try NIOAsyncChannel( + let asyncChannel = try NIOAsyncChannel( wrappingChannelSynchronously: channel, configuration: asyncChannelConfiguration ) + return HTTP1ChildConnection(asyncChannel: asyncChannel, closeFlag: closeFlag) } } /// Builds a ``ConnectionContext`` for an HTTP/1.1 request channel. static func makeHTTP1ConnectionContext( requestChannel: NIOAsyncChannel, + closeFlag: NIOLockedValueBox, peerCertificateChainFuture: EventLoopFuture? ) -> ConnectionContext { ConnectionContext( httpVersion: .http1_1, remoteAddress: try? NIOHTTPServer.SocketAddress(requestChannel.channel.remoteAddress), localAddress: try? NIOHTTPServer.SocketAddress(requestChannel.channel.localAddress), - peerCertificateChainFuture: peerCertificateChainFuture + peerCertificateChainFuture: peerCertificateChainFuture, + closeBacking: .http1_1(closeFlag: closeFlag) ) } @@ -205,8 +221,10 @@ extension NIOHTTPServer { /// The caller (the dispatcher) owns the channel's `executeThenClose`, /// so this method only iterates inbound requests and writes responses; /// it never closes the channel itself. The loop terminates when the - /// peer closes the connection, the task is cancelled, or an error - /// occurs. + /// peer closes the connection, the task is cancelled, the handler + /// signals close (which causes the ``HTTPKeepAliveHandler`` to close + /// the channel after the response — the next iterator read then returns + /// `nil`), or an error occurs. func handleHTTP1RequestLoop( inbound: NIOAsyncChannelInboundStream, outbound: NIOAsyncChannelOutboundWriter, diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+RequestContext.swift b/Sources/NIOHTTPServer/NIOHTTPServer+RequestContext.swift index bb8eb70..dadaab4 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+RequestContext.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+RequestContext.swift @@ -22,6 +22,7 @@ extension NIOHTTPServer { /// Conforms to: /// - ``HTTPServerCapability/ConnectionInfo`` — peer / local addresses. /// - ``HTTPServerCapability/PeerCertificate`` — mTLS-validated peer chain. + /// - ``HTTPServerCapability/CloseableConnection`` — `signalConnectionClose()`. /// /// Generic library code can constrain on these capabilities to access /// per-request data without depending on ``NIOHTTPServer`` directly. @@ -65,3 +66,18 @@ extension NIOHTTPServer.RequestContext: HTTPServerCapability.PeerCertificate { } } } + +@available(anyAppleOS 26.0, *) +extension NIOHTTPServer.RequestContext: HTTPServerCapability.CloseableConnection { + /// Signal that the connection should close after the current response. + /// + /// Non-blocking and idempotent. Effective after the current response: + /// + /// - On HTTP/1.1, the response carries `Connection: close` and the + /// channel is closed once the response has been written. + /// - On HTTP/2, the connection sends `GOAWAY`; in-flight streams + /// complete normally before the connection closes. + public func signalConnectionClose() { + self.connectionContext.signalConnectionClose() + } +} diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift index f5cf9e5..84ac17b 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift @@ -30,7 +30,7 @@ import X509 @available(anyAppleOS 26.0, *) extension NIOHTTPServer { typealias NegotiatedChannel = NIONegotiatedHTTPVersion< - NIOAsyncChannel, + HTTP1ChildConnection, (any Channel, NIOHTTP2Handler.AsyncStreamMultiplexer>) > @@ -91,15 +91,16 @@ extension NIOHTTPServer { } switch negotiatedChannel { - case .http1_1(let requestChannel): + case .http1_1(let child): // The dispatcher owns the channel's `executeThenClose` so the // `NIOAsyncWriter` is finished cleanly whether or not the // connection handler called `handleRequests`. do { - try await requestChannel.executeThenClose { inbound, outbound in - let chainFuture = requestChannel.channel.nioSSL_peerValidatedCertificateChain() + try await child.asyncChannel.executeThenClose { inbound, outbound in + let chainFuture = child.asyncChannel.channel.nioSSL_peerValidatedCertificateChain() let context = NIOHTTPServer.makeHTTP1ConnectionContext( - requestChannel: requestChannel, + requestChannel: child.asyncChannel, + closeFlag: child.closeFlag, peerCertificateChainFuture: chainFuture ) let connection = Connection( @@ -156,7 +157,8 @@ extension NIOHTTPServer { httpVersion: .http2, remoteAddress: try? NIOHTTPServer.SocketAddress(connectionChannel.remoteAddress), localAddress: try? NIOHTTPServer.SocketAddress(connectionChannel.localAddress), - peerCertificateChainFuture: peerCertificateChainFuture + peerCertificateChainFuture: peerCertificateChainFuture, + closeBacking: .http2(connectionChannel: connectionChannel) ) } diff --git a/Sources/NIOHTTPServer/ServerChannel.swift b/Sources/NIOHTTPServer/ServerChannel.swift index ae65104..e5c6a2e 100644 --- a/Sources/NIOHTTPServer/ServerChannel.swift +++ b/Sources/NIOHTTPServer/ServerChannel.swift @@ -22,7 +22,7 @@ extension NIOHTTPServer { /// Upgrade. enum ServerChannel { case plaintextHTTP1_1( - channel: NIOAsyncChannel, Never>, + channel: NIOAsyncChannel, quiescingHelper: ServerQuiescingHelper ) diff --git a/Tests/NIOHTTPServerTests/ConnectionLifecycleTests.swift b/Tests/NIOHTTPServerTests/ConnectionLifecycleTests.swift index 777207a..37ae183 100644 --- a/Tests/NIOHTTPServerTests/ConnectionLifecycleTests.swift +++ b/Tests/NIOHTTPServerTests/ConnectionLifecycleTests.swift @@ -139,7 +139,7 @@ struct ConnectionLifecycleTests { NIOHTTPServer.Reader, NIOHTTPServer.ResponseSender > { - HTTPServerClosureRequestHandler { request, requestContext, reader, responseSender in + HTTPServerClosureRequestHandler { _, _, reader, responseSender in try await NIOHTTPServerTests.echoResponse(readUpTo: 1024, reader: reader, sender: responseSender) } } @@ -232,11 +232,7 @@ struct ConnectionLifecycleTests { NIOHTTPServer.RequestContext, NIOHTTPServer.Reader, NIOHTTPServer.ResponseSender - > = HTTPServerClosureRequestHandler { - request, - requestContext, - reader, - responseSender in + > = HTTPServerClosureRequestHandler { _, _, _, responseSender in let arrived = arrivedCounter.withLockedValue { value -> Int in value += 1 return value @@ -293,6 +289,347 @@ struct ConnectionLifecycleTests { #expect(state.observedFinalCounter.withLockedValue { $0 } == numStreams) } + /// HTTP/1.1: a request handler that signals close has its response carry + /// `Connection: close`, and a follow-up read on the same socket returns nil. + @available(anyAppleOS 26.0, *) + @Test("signalConnectionClose() on HTTP/1.1", .timeLimit(.minutes(1))) + func testSignalConnectionCloseHTTP1_1() async throws { + let server = try NIOHTTPServerTests.makePlaintextHTTP1Server(logger: Self.serverLogger) + + let connectionHandler = NIOHTTPServerDefaultConnectionHandler( + handler: HTTPServerClosureRequestHandler { _, requestContext, _, responseSender in + requestContext.signalConnectionClose() + var buffer = UniqueArray(copying: []) + try await responseSender.sendAndFinish(.init(status: .unauthorized), buffer: &buffer) + } + ) + + try await Self.withServer(server: server, connectionHandler: connectionHandler) { serverAddress in + 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)) + + var iterator = inbound.makeAsyncIterator() + let head = try await iterator.next() + guard case .head(let response) = head else { + Issue.record("Expected response head but got \(String(describing: head))") + return + } + #expect(response.status == .unauthorized) + #expect(response.headerFields[.connection] == "close") + + while let part = try await iterator.next() { + if case .end = part { break } + } + + let afterEnd = try await iterator.next() + #expect(afterEnd == nil) + } + } + } + + /// HTTP/2: a request handler that signals close causes the connection to + /// stop accepting new streams. In-flight streams complete normally before + /// the connection is torn down. + @available(anyAppleOS 26.0, *) + @Test("signalConnectionClose() on HTTP/2", .timeLimit(.minutes(1))) + func testSignalConnectionCloseHTTP2() async throws { + let (server, serverChain) = try NIOHTTPServerTests.makeSecureUpgradeServer(logger: Self.serverLogger) + let elg: EventLoopGroup = .singletonMultiThreadedEventLoopGroup + let numInflight = 3 + let allRequestsReceived = elg.any().makePromise(of: Void.self) + let arrivedCounter = NIOLockedValueBox(0) + + let connectionHandler = NIOHTTPServerDefaultConnectionHandler( + handler: HTTPServerClosureRequestHandler { _, requestContext, _, responseSender in + let arrived = arrivedCounter.withLockedValue { value -> Int in + value += 1 + return value + } + if arrived == numInflight { + requestContext.signalConnectionClose() + allRequestsReceived.succeed() + } else { + try await allRequestsReceived.futureResult.get() + } + var buffer = UniqueArray(copying: []) + try await responseSender.sendAndFinish(.init(status: .ok), buffer: &buffer) + } + ) + try await Self.withServer(server: server, connectionHandler: connectionHandler) { serverAddress in + let clientChannel = try await ClientBootstrap(group: elg) + .connectToTestSecureUpgradeHTTPServer( + at: serverAddress, + trustRoots: serverChain.chain, + applicationProtocol: HTTPVersion.http2.alpnIdentifier + ) + guard case .http2(let streamManager) = clientChannel else { + Issue.record("Expected HTTP/2 channel, got \(clientChannel).") + return + } + + try await withThrowingTaskGroup { group in + for _ in 1...numInflight { + group.addTask { + let stream = try await streamManager.openStream() + try await stream.executeThenClose { inbound, outbound in + try await outbound.write( + .head(.init(method: .get, scheme: "https", authority: "", path: "/")) + ) + try await outbound.write(.end(nil)) + var iterator = inbound.makeAsyncIterator() + var sawHead = false + while let part = try await iterator.next() { + if case .head(let response) = part { + #expect(response.status == .ok) + sawHead = true + } + if case .end = part { break } + } + #expect(sawHead, "All in-flight streams should complete normally with status ok") + } + } + } + try await group.waitForAll() + } + } + } + + /// HTTP/1.1: calling `signalConnectionClose()` twice within a single request + /// handler is harmless — the response still carries a single + /// `Connection: close` header and the channel is closed exactly once. + @available(anyAppleOS 26.0, *) + @Test("signalConnectionClose() is idempotent (HTTP/1.1)", .timeLimit(.minutes(1))) + func testSignalConnectionCloseIdempotent() async throws { + let server = try NIOHTTPServerTests.makePlaintextHTTP1Server(logger: Self.serverLogger) + + let connectionHandler = NIOHTTPServerDefaultConnectionHandler( + handler: HTTPServerClosureRequestHandler { _, requestContext, _, responseSender in + requestContext.signalConnectionClose() + requestContext.signalConnectionClose() + requestContext.signalConnectionClose() + var buffer = UniqueArray(copying: []) + try await responseSender.sendAndFinish(.init(status: .ok), buffer: &buffer) + } + ) + + try await Self.withServer(server: server, connectionHandler: connectionHandler) { serverAddress in + 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)) + + var iterator = inbound.makeAsyncIterator() + let head = try await iterator.next() + guard case .head(let response) = head else { + Issue.record("Expected response head but got \(String(describing: head))") + return + } + + // Look for any duplicate `Connection` headers; the field should + // appear exactly once with value "close". + let connectionValues = response.headerFields[values: .connection] + #expect(connectionValues == ["close"]) + + while let part = try await iterator.next() { + if case .end = part { break } + } + + #expect(try await iterator.next() == nil) + } + } + } + + /// HTTP/1.1: calling `signalConnectionClose()` AFTER the response head is + /// on the wire. The channel still closes after the response `.end`, and the + /// head does NOT carry `Connection: close` — it can't be amended once it's + /// already been streamed. + @available(anyAppleOS 26.0, *) + @Test("signalConnectionClose() after response head is on the wire (HTTP/1.1)", .timeLimit(.minutes(1))) + func testSignalConnectionCloseAfterHead() async throws { + let server = try NIOHTTPServerTests.makePlaintextHTTP1Server(logger: Self.serverLogger) + + let (canSignalStream, canSignal) = AsyncStream.makeStream() + + let connectionHandler = NIOHTTPServerDefaultConnectionHandler( + handler: HTTPServerClosureRequestHandler { _, requestContext, _, responseSender in + let writer = try await responseSender.send(.init(status: .ok)) + // Wait for the client to confirm it has read the head so + // signalConnectionClose() is deterministically after the head + // has reached the wire. + var iterator = canSignalStream.makeAsyncIterator() + _ = await iterator.next() + requestContext.signalConnectionClose() + var buffer = UniqueArray(copying: [0x21]) + try await writer.finish(buffer: &buffer, finalElement: nil) + } + ) + + try await Self.withServer(server: server, connectionHandler: connectionHandler) { serverAddress in + 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)) + + var iterator = inbound.makeAsyncIterator() + let head = try await iterator.next() + guard case .head(let response) = head else { + Issue.record("Expected response head but got \(String(describing: head))") + return + } + #expect(response.status == .ok) + // The head reached the wire before signalConnectionClose ran, + // so it can't have been amended. + #expect( + response.headerFields[.connection] == nil, + "Head was already on the wire when close was signalled — can't be amended; got \(response.headerFields)" + ) + + // Unblock the server to signal close and finish the response. + canSignal.yield() + canSignal.finish() + + while let part = try await iterator.next() { + if case .end = part { break } + } + + #expect(try await iterator.next() == nil) + } + } + } + + /// HTTP/1.1: calling `signalConnectionClose()` mid-response — after the + /// response head has already been streamed to the wire — still closes the + /// channel once response `.end` is written. Server-side timeouts are + /// disabled on this server, so the ONLY thing that can close the channel + /// is `HTTPKeepAliveHandler`'s own close path on response `.end`. Without + /// the mid-response check on `closeSignalled`, this test hangs and hits + /// the 1-minute time limit. + @available(anyAppleOS 26.0, *) + @Test("signalConnectionClose() mid-response after body drained (HTTP/1.1)", .timeLimit(.minutes(1))) + func testSignalConnectionCloseMidResponseAfterDrain() 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 (canSignalStream, canSignal) = AsyncStream.makeStream() + + let connectionHandler = NIOHTTPServerDefaultConnectionHandler( + handler: HTTPServerClosureRequestHandler { _, requestContext, reader, responseSender in + // Drain the body so `invokeHandler` recovers the iterator and the + // request loop stays alive — otherwise the loop exits when the + // handler returns, and the dispatcher closes the channel anyway. + var body = UniqueArray() + body.reserveCapacity(1024) + _ = try await reader.collect(into: &body) + + let writer = try await responseSender.send(.init(status: .ok)) + + // Wait for the client to confirm it has read the head. + var iterator = canSignalStream.makeAsyncIterator() + _ = await iterator.next() + + // Signal close AFTER the head is on the wire. + requestContext.signalConnectionClose() + + var buffer = UniqueArray(copying: [0x21]) + try await writer.finish(buffer: &buffer, finalElement: nil) + } + ) + + try await Self.withServer(server: server, connectionHandler: connectionHandler) { serverAddress in + 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)) + + var iterator = inbound.makeAsyncIterator() + let head = try await iterator.next() + guard case .head = head else { + Issue.record("Expected response head") + return + } + canSignal.yield() + canSignal.finish() + + while let part = try await iterator.next() { + if case .end = part { break } + } + + #expect(try await iterator.next() == nil) + } + } + } + + /// HTTP/2: after a stream signals close, the connection drains existing + /// streams and rejects new stream creation. Opening a stream against the + /// already-quiesced connection fails (NIOHTTP2's GOAWAY mechanics). + /// TODO: this test is disabled since it's currently flaky; because of a bug in NIO where inbound NIOAsyncChannels are + /// dropped if not consumed, and a precondition fails because they've not been finished. + @available(anyAppleOS 26.0, *) + @Test("HTTP/2 rejects new streams after signalConnectionClose()", .disabled()) + func testHTTP2RejectsNewStreamsAfterClose() async throws { + let (server, serverChain) = try NIOHTTPServerTests.makeSecureUpgradeServer(logger: Self.serverLogger) + let elg: EventLoopGroup = .singletonMultiThreadedEventLoopGroup + + let connectionHandler = NIOHTTPServerDefaultConnectionHandler( + handler: HTTPServerClosureRequestHandler { _, requestContext, _, responseSender in + requestContext.signalConnectionClose() + var buffer = UniqueArray(copying: []) + try await responseSender.sendAndFinish(.init(status: .ok), buffer: &buffer) + } + ) + + try await Self.withServer(server: server, connectionHandler: connectionHandler) { serverAddress in + let clientChannel = try await ClientBootstrap(group: elg) + .connectToTestSecureUpgradeHTTPServer( + at: serverAddress, + trustRoots: serverChain.chain, + applicationProtocol: HTTPVersion.http2.alpnIdentifier + ) + guard case .http2(let streamManager) = clientChannel else { + Issue.record("Expected HTTP/2 channel, got \(clientChannel).") + return + } + + // First stream completes normally and signals connection close. + let stream = try await streamManager.openStream() + try await stream.executeThenClose { inbound, outbound in + try await outbound.write(.head(.init(method: .get, scheme: "https", authority: "", path: "/"))) + try await outbound.write(.end(nil)) + var iterator = inbound.makeAsyncIterator() + while let part = try await iterator.next() { + if case .end = part { break } + } + } + + // After the close signal has been observed by the connection-management + // handler, opening a new stream should fail. We poll for a short + // window since the GOAWAY processing is asynchronous to the response + // we just observed. + var rejected = false + for _ in 0..<50 { + do { + _ = try await streamManager.openStream() + } catch { + rejected = true + break + } + try await Task.sleep(for: .milliseconds(50)) + } + #expect(rejected, "Expected new-stream creation to fail after GOAWAY.") + } + } + /// A throwing connection handler is logged at debug level by the server but /// doesn't bring it down: a subsequent connection on the same server is /// served normally. diff --git a/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1.swift b/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1_1.swift similarity index 95% rename from Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1.swift rename to Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1_1.swift index 9efd034..fc11b8c 100644 --- a/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1.swift +++ b/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1_1.swift @@ -33,7 +33,7 @@ extension NIOHTTPServer { { // The server requires a NIOAsyncChannel, so we create one from the test channel let serverTestAsyncChannel = try await testChannel.eventLoop.submit { - try NIOAsyncChannel, Never>( + try NIOAsyncChannel( wrappingChannelSynchronously: testChannel, configuration: .init() ) diff --git a/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/TestingChannelServer+HTTP1.swift b/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/TestingChannelServer+HTTP1_1.swift similarity index 100% rename from Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/TestingChannelServer+HTTP1.swift rename to Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/TestingChannelServer+HTTP1_1.swift