From 634287f241454d364f8a5599970d842ff5689d8e Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Wed, 1 Apr 2026 13:14:57 +0100 Subject: [PATCH 1/2] Improve error handling Motivation: The `serveInsecureHTTP1_1` and `serveSecureUpgrade` methods listen for incoming activity on the server channel and dispatch connections/requests to child tasks in order to process concurrently. However, these methods use a `withThrowingDiscardingTaskGroup`, and call many `throw`ing functions from within the task group. Since a `discarding` task group implicitly cancels *all* child tasks as soon as an error is encountered in *any* child task, we should carefully reason about and handle all errors that could arise within the task group. Modifications: - Removed the redundant `throw` keyword from the `handleRequestChannel` method declaration. - Refactored `serveInsecureHTTP1_1` and `serveSecureUpgrade` to handle errors more explicitly without propagating them up to the parent task group. As a result, it was then possible to replace `withThrowingDiscardingTaskGroup` with `withDiscardingTaskGroup`. Result: Better error handling. --- .../NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift | 34 +++- .../NIOHTTPServer+SecureUpgrade.swift | 138 +++++++++++----- Sources/NIOHTTPServer/NIOHTTPServer.swift | 147 ++++++++++-------- 3 files changed, 211 insertions(+), 108 deletions(-) diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift index 8cd4a0d..764259c 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift @@ -22,21 +22,39 @@ import NIOPosix @available(macOS 26.2, iOS 26.2, watchOS 26.2, tvOS 26.2, visionOS 26.2, *) extension NIOHTTPServer { + /// Serves incoming plaintext HTTP/1.1 connections. + /// + /// Each connection is handled concurrently in its own child task. Individual connection errors are handled within + /// the child tasks and do not affect other connections. + /// + /// - Parameters: + /// - serverChannel: The async channel that produces incoming HTTP/1.1 connections. + /// - handler: The request handler. + /// + /// - Throws: If an error occurs while iterating the incoming connection stream. func serveInsecureHTTP1_1( serverChannel: NIOAsyncChannel, Never>, handler: some HTTPServerRequestHandler ) async throws { - try await withThrowingDiscardingTaskGroup { group in - try await serverChannel.executeThenClose { inbound in - for try await http1Channel in inbound { - group.addTask { - try await self.handleRequestChannel( - channel: http1Channel, - handler: handler - ) + try await serverChannel.executeThenClose { inbound in + let inboundConnectionIterationError = await withDiscardingTaskGroup { group -> (any Error)? in + do { + for try await http1Channel in inbound { + group.addTask { + await self.handleRequestChannel(channel: http1Channel, handler: handler) + } } + + return nil + } catch { + return error } } + + if let inboundConnectionIterationError { + // The error occurred while iterating the inbound connection stream + throw inboundConnectionIterationError + } } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift index 65657f7..17433ac 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift @@ -35,50 +35,116 @@ extension NIOHTTPServer { (any Channel, NIOHTTP2Handler.AsyncStreamMultiplexer>) > + /// Serves incoming connections. Each connection undergoes ALPN negotiation to determine whether to use HTTP/1.1 or + /// HTTP/2, and requests are then handled over the negotiated protocol. + /// + /// Each accepted connection is handled concurrently in its own child task. Individual negotiation errors and + /// connection errors are handled within the child tasks and do not affect other connections. + /// + /// - Parameters: + /// - serverChannel: The async channel that produces incoming connections. + /// - handler: The request handler. + /// + /// - Throws: If an error occurs while iterating the incoming connection stream. func serveSecureUpgrade( serverChannel: NIOAsyncChannel, Never>, handler: some HTTPServerRequestHandler ) async throws { - try await withThrowingDiscardingTaskGroup { group in - try await serverChannel.executeThenClose { inbound in - for try await upgradeResult in inbound { - group.addTask { - do { - try await withThrowingDiscardingTaskGroup { connectionGroup in - switch try await upgradeResult.get() { - case .http1_1(let http1Channel): - let chainFuture = http1Channel.channel.nioSSL_peerValidatedCertificateChain() - Self.$connectionContext.withValue(ConnectionContext(chainFuture)) { - connectionGroup.addTask { - try await self.handleRequestChannel( - channel: http1Channel, - handler: handler - ) - } - } - case .http2((let http2Connection, let http2Multiplexer)): - do { - let chainFuture = http2Connection.nioSSL_peerValidatedCertificateChain() - try await Self.$connectionContext.withValue(ConnectionContext(chainFuture)) { - for try await http2StreamChannel in http2Multiplexer.inbound { - connectionGroup.addTask { - try await self.handleRequestChannel( - channel: http2StreamChannel, - handler: handler - ) - } - } - } - } catch { - self.logger.debug("HTTP2 connection closed: \(error)") - } - } + try await serverChannel.executeThenClose { inbound in + let inboundConnectionIterationError = await withDiscardingTaskGroup { connectionGroup -> (any Error)? in + do { + for try await upgradeResult in inbound { + connectionGroup.addTask { + let negotiatedChannel: NegotiatedChannel + + do { + negotiatedChannel = try await upgradeResult.get() + } catch { + self.logger.debug("Negotiating ALPN failed", metadata: ["error": "\(error)"]) + return } - } catch { - self.logger.debug("Negotiating ALPN failed: \(error)") + + switch negotiatedChannel { + case .http1_1(let requestChannel): + await self.serveHTTP1Connection( + requestChannel: requestChannel, + handler: handler + ) + + case .http2((let connectionChannel, let multiplexer)): + await self.serveHTTP2Connection( + connectionChannel: connectionChannel, + multiplexer: multiplexer, + handler: handler + ) + } + } + } + + return nil + } catch { + return error + } + } + + if let inboundConnectionIterationError { + // The error occurred while iterating the inbound connection stream + throw inboundConnectionIterationError + } + } + } + + /// Serves a HTTP/1.1 connection. + /// + /// - Parameters: + /// - requestChannel: The HTTP/1.1 request channel. + /// - handler: The request handler. + private func serveHTTP1Connection( + requestChannel: NIOAsyncChannel, + handler: some HTTPServerRequestHandler + ) async { + let chainFuture = requestChannel.channel.nioSSL_peerValidatedCertificateChain() + + await Self.$connectionContext.withValue(ConnectionContext(chainFuture)) { + await self.handleRequestChannel( + channel: requestChannel, + handler: handler + ) + } + } + + /// Serves a HTTP/2 connection by iterating the stream channels and handling each stream concurrently. + /// + /// - Note: Stream iteration errors are logged but do not propagate to the caller. + /// + /// - Parameters: + /// - connectionChannel: The underlying NIO channel for the HTTP/2 connection. + /// - multiplexer: The HTTP/2 stream multiplexer. + /// - handler: The request handler. + private func serveHTTP2Connection( + connectionChannel: any Channel, + multiplexer: NIOHTTP2Handler.AsyncStreamMultiplexer>, + handler: some HTTPServerRequestHandler + ) async { + await withDiscardingTaskGroup { streamGroup in + do { + let chainFuture = connectionChannel.nioSSL_peerValidatedCertificateChain() + + try await Self.$connectionContext.withValue(ConnectionContext(chainFuture)) { + for try await streamChannel in multiplexer.inbound { + streamGroup.addTask { + await self.handleRequestChannel( + channel: streamChannel, + handler: handler + ) } } } + } catch { + self.logger.error( + "Error thrown while iterating over incoming HTTP/2 streams", + metadata: ["error": "\(error)"] + ) } } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer.swift b/Sources/NIOHTTPServer/NIOHTTPServer.swift index fb66940..96755c6 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer.swift @@ -213,82 +213,101 @@ public struct NIOHTTPServer: HTTPServer { } } + /// Handles a single HTTP request. + /// + /// - Note: Errors do not propagate to the caller. When an error occurs, it is logged and the channel is closed. + /// + /// - Parameters: + /// - channel: The async channel to read the request from and write the response to. + /// - handler: The request handler. func handleRequestChannel( channel: NIOAsyncChannel, handler: some HTTPServerRequestHandler - ) async throws { + ) async { do { - try await channel - .executeThenClose { inbound, outbound in - var iterator = inbound.makeAsyncIterator() - - let httpRequest: HTTPRequest - switch try await iterator.next() { - case .head(let request): - httpRequest = request - case .body: - self.logger.debug("Unexpectedly received body on connection. Closing now") - outbound.finish() - return - case .end: - self.logger.debug("Unexpectedly received end on connection. Closing now") - outbound.finish() - return - case .none: - self.logger.trace("No more requests parts on connection") - return - } + try await channel.executeThenClose { inbound, outbound in + var iterator = inbound.makeAsyncIterator() + + let nextPart: HTTPRequestPart? + do { + nextPart = try await iterator.next() + } catch { + self.logger.error( + "Error thrown while advancing the request iterator", + metadata: ["error": "\(error)"] + ) + throw error + } - let readerState = HTTPRequestConcludingAsyncReader.ReaderState() - let writerState = HTTPResponseConcludingAsyncWriter.WriterState() - - do { - try await handler.handle( - request: httpRequest, - requestContext: HTTPRequestContext(), - requestBodyAndTrailers: HTTPRequestConcludingAsyncReader( - iterator: iterator, - readerState: readerState - ), - responseSender: HTTPResponseSender { response in - try await outbound.write(.head(response)) - return HTTPResponseConcludingAsyncWriter( - writer: outbound, - writerState: writerState - ) - } sendInformational: { response in - try await outbound.write(.head(response)) - } - ) - } catch { - logger.error("Error thrown while handling connection: \(error)") - if !readerState.wrapped.withLock({ $0.finishedReading }) { - logger.error("Did not finish reading but error thrown.") - // TODO: if h2 reset stream; if h1 try draining request? - } - if !writerState.wrapped.withLock({ $0.finishedWriting }) { - logger.error("Did not write response but error thrown.") - // TODO: we need to do something, possibly just close the connection or - // reset the stream with the appropriate error. + let httpRequest: HTTPRequest + switch nextPart { + case .head(let request): + httpRequest = request + case .body: + self.logger.debug("Unexpectedly received body on connection. Closing now") + outbound.finish() + return + case .end: + self.logger.debug("Unexpectedly received end on connection. Closing now") + outbound.finish() + return + case .none: + self.logger.trace("No more requests parts on connection") + return + } + + let readerState = HTTPRequestConcludingAsyncReader.ReaderState() + let writerState = HTTPResponseConcludingAsyncWriter.WriterState() + + do { + try await handler.handle( + request: httpRequest, + requestContext: HTTPRequestContext(), + requestBodyAndTrailers: HTTPRequestConcludingAsyncReader( + iterator: iterator, + readerState: readerState + ), + responseSender: HTTPResponseSender { response in + try await outbound.write(.head(response)) + return HTTPResponseConcludingAsyncWriter( + writer: outbound, + writerState: writerState + ) + } sendInformational: { response in + try await outbound.write(.head(response)) } - throw error + ) + } catch { + if !readerState.wrapped.withLock({ $0.finishedReading }) { + self.logger.error("Did not finish reading but error thrown.") + // TODO: if h2 reset stream; if h1 try draining request? } - // TODO: handle other state scenarios. - // For example, if we're using h2 and we didn't finish reading but we wrote back - // a response, we should send a RST_STREAM with NO_ERROR set. - // If we finished reading but we didn't write back a response, then RST_STREAM - // is also likely appropriate but unclear about the error. - // For h1, we should close the connection. + if !writerState.wrapped.withLock({ $0.finishedWriting }) { + self.logger.error("Did not write response but error thrown.") + // TODO: we need to do something, possibly just close the connection or + // reset the stream with the appropriate error. + } - // Finish the outbound and wait on the close future to make sure all pending - // writes are actually written. - outbound.finish() - try await channel.channel.closeFuture.get() + throw error } + + // TODO: handle other state scenarios. + // For example, if we're using h2 and we didn't finish reading but we wrote back + // a response, we should send a RST_STREAM with NO_ERROR set. + // If we finished reading but we didn't write back a response, then RST_STREAM + // is also likely appropriate but unclear about the error. + // For h1, we should close the connection. + + // Finish the outbound and wait on the close future to make sure all pending + // writes are actually written. + outbound.finish() + try await channel.channel.closeFuture.get() + } } catch { - self.logger.debug("Error thrown while handling connection: \(error)") // TODO: We need to send a response head here potentially + self.logger.error("Error thrown while handling connection", metadata: ["error": "\(error)"]) + try? await channel.channel.close() } } From cb84c52925353d9ba130fa9d168f8dfd184fd399 Mon Sep 17 00:00:00 2001 From: Aryan Shah Date: Mon, 13 Apr 2026 23:04:47 +0100 Subject: [PATCH 2/2] Add comment explaining use of `withDiscardingTaskGroup` --- Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift | 4 ++++ Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift index 764259c..d8c2f01 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift @@ -37,6 +37,10 @@ extension NIOHTTPServer { handler: some HTTPServerRequestHandler ) async throws { try await serverChannel.executeThenClose { inbound in + // We don't use a `withThrowingDiscardingTaskGroup` here because an error thrown from the body or a child + // task would immediately propagate upwards, cancelling all child tasks and bringing down the entire server. + // We instead use a non-throwing discarding task group so that errors in the body (e.g. from iterating + // `inbound`) must be caught and handled directly. let inboundConnectionIterationError = await withDiscardingTaskGroup { group -> (any Error)? in do { for try await http1Channel in inbound { diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift index 17433ac..3b708f0 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift @@ -51,6 +51,10 @@ extension NIOHTTPServer { handler: some HTTPServerRequestHandler ) async throws { try await serverChannel.executeThenClose { inbound in + // We don't use a `withThrowingDiscardingTaskGroup` here because an error thrown from the body or a child + // task would immediately propagate upwards, cancelling all child tasks and bringing down the entire server. + // We instead use a non-throwing discarding task group so that errors in the body (e.g. from iterating + // `inbound`) must be caught and handled directly. let inboundConnectionIterationError = await withDiscardingTaskGroup { connectionGroup -> (any Error)? in do { for try await upgradeResult in inbound {