From e32137f48753b188b63f44bfd215c43c024efc52 Mon Sep 17 00:00:00 2001 From: Gus Cairo Date: Fri, 9 Oct 2026 16:18:42 +0100 Subject: [PATCH 1/2] Support network path change notifications --- .../ConnectionEventsHandler.swift | 45 +++ ...verCapability+ConnectionCapabilities.swift | 16 + .../NIOHTTPServer+ConnectionContext.swift | 41 +- .../NIOHTTPServer+ConnectionEvents.swift | 160 ++++++++ .../NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift | 54 ++- .../NIOHTTPServer/NIOHTTPServer+HTTP3.swift | 87 ++-- .../NIOHTTPServer+RequestContext.swift | 15 +- .../NIOHTTPServer+SecureUpgrade.swift | 50 +-- Sources/NIOHTTPServer/NIOHTTPServer.swift | 11 + .../ConnectionEventsTests.swift | 382 ++++++++++++++++++ .../NIOHTTPServer+ServiceLifecycleTests.swift | 1 + .../NIOHTTPServer+HTTP1.swift | 2 +- 12 files changed, 777 insertions(+), 87 deletions(-) create mode 100644 Sources/NIOHTTPServer/ConnectionEventsHandler.swift create mode 100644 Sources/NIOHTTPServer/NIOHTTPServer+ConnectionEvents.swift create mode 100644 Tests/NIOHTTPServerTests/ConnectionEventsTests.swift diff --git a/Sources/NIOHTTPServer/ConnectionEventsHandler.swift b/Sources/NIOHTTPServer/ConnectionEventsHandler.swift new file mode 100644 index 0000000..582f9fa --- /dev/null +++ b/Sources/NIOHTTPServer/ConnectionEventsHandler.swift @@ -0,0 +1,45 @@ +//===----------------------------------------------------------------------===// +// +// 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 + +/// This channel handler keeps a connection's ``NIOHTTPServer/ConnectionContext`` current with the events fired into the +/// connection channel's pipeline, and reports them through the context's +/// ``NIOHTTPServer/ConnectionContext/connectionEvents``, which it finishes once the connection closes. +@available(anyAppleOS 27.0, *) +final class ConnectionEventsHandler: ChannelInboundHandler, RemovableChannelHandler { + typealias InboundIn = NIOAny + typealias InboundOut = NIOAny + + private let connectionContext: NIOHTTPServer.ConnectionContext + + init(connectionContext: NIOHTTPServer.ConnectionContext) { + self.connectionContext = connectionContext + } + + func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) { + // TODO: Translate the QUIC stack's own events (for example, the peer migrating the connection to a new path, or + // a new path being validated) into `ConnectionEvent`s, once it fires them into the connection channel. + if let event = event as? NIOHTTPServer.ConnectionEvent { + self.connectionContext.connectionEventReceived(event) + } + + context.fireUserInboundEventTriggered(event) + } + + func channelInactive(context: ChannelHandlerContext) { + self.connectionContext.unsubscribeFromConnectionEvents() + context.fireChannelInactive() + } +} diff --git a/Sources/NIOHTTPServer/HTTPServerCapability+ConnectionCapabilities.swift b/Sources/NIOHTTPServer/HTTPServerCapability+ConnectionCapabilities.swift index e33bdfb..4802848 100644 --- a/Sources/NIOHTTPServer/HTTPServerCapability+ConnectionCapabilities.swift +++ b/Sources/NIOHTTPServer/HTTPServerCapability+ConnectionCapabilities.swift @@ -25,6 +25,9 @@ extension HTTPServerCapability { /// underlying transport cannot report an address. public protocol ConnectionInfo: RequestContext { /// The peer's address, when known. + /// + /// This can change during the connection's lifetime: for example, when a client migrates an HTTP/3 connection + /// to a new network path. var remoteAddress: NIOHTTPServer.SocketAddress? { get } /// The local address the connection is bound to, when known. @@ -41,4 +44,17 @@ extension HTTPServerCapability { /// The peer's validated certificate chain, when available. var validatedPeerCertificateChain: X509.ValidatedCertificateChain? { get } } + + /// A request-context capability exposing the events that happen on the connection carrying the request. + /// + /// Servers whose request context conforms to this capability let request + /// handlers react to changes in the connection beneath them, such as the + /// peer's address changing when a client migrates an HTTP/3 connection to + /// a new network path. + public protocol ConnectionEvents: RequestContext { + /// The events that happen on the connection carrying the request. + /// + /// The sequence finishes once the connection closes. + var connectionEvents: NIOHTTPServer.ConnectionEvents { get } + } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift b/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift index 7b3dd41..b5bef8f 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 @@ -36,11 +37,12 @@ extension NIOHTTPServer { /// /// Carries connection-scoped data such as the negotiated HTTP version, the /// peer / local addresses, and the peer's validated certificate chain (when - /// applicable). + /// applicable), along with the events that happen on the connection. /// /// User code accesses this state via the corresponding ``RequestContext`` /// capabilities (``HTTPServerCapability/ConnectionInfo``, - /// ``HTTPServerCapability/PeerCertificate``) when handling individual + /// ``HTTPServerCapability/PeerCertificate``, + /// ``HTTPServerCapability/ConnectionEvents``) when handling individual /// requests, and directly when implementing an /// ``NIOHTTPServerConnectionHandler``. public struct ConnectionContext: Sendable { @@ -48,7 +50,12 @@ extension NIOHTTPServer { public let httpVersion: HTTPVersion /// The peer's address, when known. - public let remoteAddress: NIOHTTPServer.SocketAddress? + /// + /// Over HTTP/3 this can change during the connection's lifetime, when the client migrates the connection to a + /// new network path. ``connectionEvents`` reports each change. + public var remoteAddress: NIOHTTPServer.SocketAddress? { + self.currentRemoteAddress.withLockedValue { $0 } + } /// The local address the connection is bound to, when known. public let localAddress: NIOHTTPServer.SocketAddress? @@ -58,6 +65,15 @@ extension NIOHTTPServer { /// derived validated chain. public var validatedPeerCertificateChain: X509.ValidatedCertificateChain? + /// The events that happen on this connection, such as the peer's address changing. + /// + /// The sequence finishes once the connection closes. See ``ConnectionEvents``. + public let connectionEvents: ConnectionEvents + + /// Shared by every copy of this context, so that the connection's ``ConnectionEventsHandler`` keeps them all + /// current. + private let currentRemoteAddress: NIOLockedValueBox + init( httpVersion: HTTPVersion, remoteAddress: NIOHTTPServer.SocketAddress? = nil, @@ -65,9 +81,26 @@ extension NIOHTTPServer { validatedPeerCertificateChain: X509.ValidatedCertificateChain? = nil ) { self.httpVersion = httpVersion - self.remoteAddress = remoteAddress + self.currentRemoteAddress = NIOLockedValueBox(remoteAddress) self.localAddress = localAddress self.validatedPeerCertificateChain = validatedPeerCertificateChain + self.connectionEvents = ConnectionEvents(broadcaster: ConnectionEventBroadcaster()) + } + + /// Applies an `event` received on the connection to this context, then delivers it to every iterator of + /// ``connectionEvents``. + func connectionEventReceived(_ event: ConnectionEvent) { + switch event { + case .remoteAddressChanged(let address): + self.currentRemoteAddress.withLockedValue { $0 = address } + } + + self.connectionEvents.broadcaster.yield(event) + } + + /// Ends every iterator of ``connectionEvents``, now that the connection has closed. + func unsubscribeFromConnectionEvents() { + self.connectionEvents.broadcaster.finish() } } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionEvents.swift b/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionEvents.swift new file mode 100644 index 0000000..ffcf559 --- /dev/null +++ b/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionEvents.swift @@ -0,0 +1,160 @@ +//===----------------------------------------------------------------------===// +// +// 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 Synchronization + +@available(anyAppleOS 27.0, *) +extension NIOHTTPServer { + /// Something that happened on a connection, as reported by ``ConnectionEvents``. + @nonexhaustive + public enum ConnectionEvent: Sendable, Hashable { + /// The peer's address changed. + /// + /// Over HTTP/3, a client can migrate its QUIC connection to a new network path (RFC 9000 ยง 9): for example, + /// when it moves from Wi-Fi to cellular, or when a NAT rebinds the port it is reachable on. The associated value + /// is the peer's new address, which ``ConnectionContext/remoteAddress`` reports from then on. + case remoteAddressChanged(NIOHTTPServer.SocketAddress) + } + + /// The events that happen on a connection. + /// + /// Any number of tasks can iterate this sequence concurrently: for example, a ``NIOHTTPServerConnectionHandler`` + /// alongside the request handlers of every request in flight on the connection. Each iterator receives every event + /// that happens after it is created, and finishes once the connection closes. + /// + /// An iterator that falls behind keeps only the most recent events, dropping older ones. The connection's state, + /// such as ``ConnectionContext/remoteAddress``, is always current, so read it rather than relying on receiving every + /// event. + /// + /// ```swift + /// for await event in requestContext.connectionEvents { + /// switch event { + /// case .remoteAddressChanged(let address): + /// logger.info("Client migrated", metadata: ["address": "\(address)"]) + /// default: + /// break + /// } + /// } + /// ``` + public struct ConnectionEvents: AsyncSequence, Sendable { + let broadcaster: ConnectionEventBroadcaster + + init(broadcaster: ConnectionEventBroadcaster) { + self.broadcaster = broadcaster + } + + public func makeAsyncIterator() -> AsyncIterator { + AsyncIterator(base: self.broadcaster.subscribe().makeAsyncIterator()) + } + + /// An iterator over a connection's events. + public struct AsyncIterator: AsyncIteratorProtocol { + private var base: AsyncStream.Iterator + + init(base: AsyncStream.Iterator) { + self.base = base + } + + /// Returns the next event, or `nil` once the connection has closed or the current task is cancelled. + public mutating func next(isolation actor: isolated (any Actor)?) async -> ConnectionEvent? { + await self.base.next(isolation: actor) + } + + /// Returns the next event, or `nil` once the connection has closed or the current task is cancelled. + public mutating func next() async -> ConnectionEvent? { + await self.base.next(isolation: #isolation) + } + } + } +} + +@available(*, unavailable) +extension NIOHTTPServer.ConnectionEvents.AsyncIterator: Sendable {} + +/// Delivers each of a connection's events to every ``NIOHTTPServer/ConnectionEvents`` iterator that exists when it +/// happens. +/// +/// This is what lets several handlers iterate a connection's events concurrently: the iterators of a single +/// `AsyncStream` would instead compete for its elements, with each event reaching only one of them. +@available(anyAppleOS 27.0, *) +final class ConnectionEventBroadcaster: Sendable { + /// How many undelivered events each iterator buffers before dropping the oldest. + static let bufferedEventLimit = 16 + + private struct State { + var subscribers: [Int: AsyncStream.Continuation] = [:] + var nextSubscriberID = 0 + var isFinished = false + } + + private let state = Mutex(State()) + + /// The number of iterators currently receiving events. + var subscriberCount: Int { + self.state.withLock { $0.subscribers.count } + } + + /// Returns a stream of the events that happen from now on, which finishes once ``finish()`` is called. + func subscribe() -> AsyncStream { + let (stream, continuation) = AsyncStream.makeStream( + of: NIOHTTPServer.ConnectionEvent.self, + bufferingPolicy: .bufferingNewest(Self.bufferedEventLimit) + ) + + let subscriberID: Int? = self.state.withLock { state in + guard !state.isFinished else { return nil } + let subscriberID = state.nextSubscriberID + state.nextSubscriberID += 1 + state.subscribers[subscriberID] = continuation + return subscriberID + } + + guard let subscriberID else { + // The connection has already closed, so there is nothing left to report. + continuation.finish() + return stream + } + + // Stop delivering to an iterator once it is dropped, or the task iterating it is cancelled. + continuation.onTermination = { [weak self] _ in + self?.state.withLock { _ = $0.subscribers.removeValue(forKey: subscriberID) } + } + + return stream + } + + /// Delivers `event` to every subscribed iterator. + func yield(_ event: NIOHTTPServer.ConnectionEvent) { + self.state.withLock { state in + for subscriber in state.subscribers.values { + subscriber.yield(event) + } + } + } + + /// Ends every subscribed iterator, and any that subscribe afterwards. + func finish() { + let subscribers = self.state.withLock { state in + state.isFinished = true + let subscribers = state.subscribers + state.subscribers = [:] + return subscribers + } + + // Outside the lock: finishing a stream runs its termination handler, which takes the lock. + for subscriber in subscribers.values { + subscriber.finish() + } + } +} diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift index 192c831..1d18cc0 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift @@ -34,7 +34,7 @@ extension NIOHTTPServer { /// /// - Throws: If an error occurs while iterating the incoming connection stream. func serveInsecureHTTP1_1( - connectionStream: NIOAsyncChannelInboundStream, + connectionStream: NIOAsyncChannelInboundStream, connectionHandler: Handler ) async throws { // We don't use a `withThrowingDiscardingTaskGroup` here because an error thrown from the body or a child @@ -46,8 +46,9 @@ extension NIOHTTPServer { for try await connectionChannel in connectionStream { group.addTask { await self.dispatchPlaintextHTTP1_1Connection( - requestChannel: connectionChannel.channel, - clientClosed: connectionChannel.clientClosed, + requestChannel: connectionChannel.requestChannel.channel, + clientClosed: connectionChannel.requestChannel.clientClosed, + context: connectionChannel.context, connectionHandler: connectionHandler ) } @@ -65,9 +66,9 @@ extension NIOHTTPServer { } } - /// Builds the per-connection ``Connection`` and ``ConnectionContext`` for a - /// plaintext HTTP/1.1 child channel and dispatches to the connection - /// handler. Errors from the connection handler are logged. + /// Builds the per-connection ``Connection`` for a plaintext HTTP/1.1 child + /// channel and dispatches it, alongside its ``ConnectionContext``, to the + /// connection handler. Errors from the connection handler are logged. /// /// The dispatcher owns the channel's `executeThenClose` so the /// `NIOAsyncWriter` is finished cleanly whether or not the connection @@ -75,16 +76,11 @@ extension NIOHTTPServer { private func dispatchPlaintextHTTP1_1Connection( requestChannel: sending NIOAsyncChannel, clientClosed: AsyncStream, + context: ConnectionContext, connectionHandler: Handler ) async { do { try await requestChannel.executeThenClose { inbound, outbound in - let context = ConnectionContext( - httpVersion: .plaintextHTTP1_1, - remoteAddress: try? NIOHTTPServer.SocketAddress(requestChannel.channel.remoteAddress), - localAddress: try? NIOHTTPServer.SocketAddress(requestChannel.channel.localAddress), - validatedPeerCertificateChain: nil - ) let connection = Connection( server: self, context: context, @@ -136,12 +132,21 @@ extension NIOHTTPServer { } } - /// Configures the HTTP/1.1 server pipeline and the keep-alive handler. + /// Configures the HTTP/1.1 server pipeline and the keep-alive handler, and builds the connection's + /// ``ConnectionContext``. func setupHTTP1_1Connection( channel: any Channel, isSecure: Bool - ) -> EventLoopFuture { + ) -> EventLoopFuture { channel.pipeline.configureHTTPServerPipeline().flatMapThrowing { + // Over TLS this runs once ALPN has negotiated HTTP/1.1, by which point the handshake has validated any + // peer certificate chain. + let context = ConnectionContext( + httpVersion: isSecure ? .http1_1 : .plaintextHTTP1_1, + remoteAddress: try? NIOHTTPServer.SocketAddress(channel.remoteAddress), + localAddress: try? NIOHTTPServer.SocketAddress(channel.localAddress), + validatedPeerCertificateChain: isSecure ? channel.extractPeerCertificateChain(logger: self.logger) : nil + ) try channel.pipeline.syncOperations.addHandler(HTTP1ToHTTPServerCodec(secure: isSecure)) try channel.pipeline.syncOperations.addHandler(HTTPKeepAliveHandler()) @@ -158,15 +163,20 @@ extension NIOHTTPServer { ClientClosedMonitor(clientClosed: clientClosedContinuation) ) - return HTTPRequestChannelAndCancellationSignal( - channel: try NIOAsyncChannel( - wrappingChannelSynchronously: channel, - configuration: .init( - backPressureStrategy: .init(self.configuration.backpressureStrategy), - isOutboundHalfClosureEnabled: true - ) + try channel.pipeline.syncOperations.addHandler(ConnectionEventsHandler(connectionContext: context)) + + return HTTP1ConnectionChannel( + requestChannel: HTTPRequestChannelAndCancellationSignal( + channel: try NIOAsyncChannel( + wrappingChannelSynchronously: channel, + configuration: .init( + backPressureStrategy: .init(self.configuration.backpressureStrategy), + isOutboundHalfClosureEnabled: true + ) + ), + clientClosed: clientClosed ), - clientClosed: clientClosed + context: context ) } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP3.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP3.swift index f7a0e37..f82c940 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP3.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP3.swift @@ -44,8 +44,17 @@ extension NIOHTTPServer { #endif } + /// An inbound HTTP/3 connection, and the context built for it when its pipeline was set up. + /// + /// The context travels with the connection because the connection channel's ``ConnectionEventsHandler`` keeps + /// that same context current. + struct HTTP3InboundConnection: Sendable { + var connection: HTTP3ServerConnection + var context: ConnectionContext + } + func serveHTTP3( - connectionMultiplexer: HTTP3ServerConnectionMultiplexer, + inboundConnections: AsyncStream, connectionHandler: Handler ) async { // We don't use a `withThrowingDiscardingTaskGroup` here because an error thrown from the body or a child task @@ -53,31 +62,26 @@ extension NIOHTTPServer { // instead use a non-throwing discarding task group so that errors in the body must be caught and handled // directly. await withDiscardingTaskGroup { connectionGroup in - for await connection in connectionMultiplexer.inboundConnections { + for await inboundConnection in inboundConnections { connectionGroup.addTask { - await self.dispatchHTTP3Connection(connection, handler: connectionHandler) + await self.dispatchHTTP3Connection(inboundConnection, handler: connectionHandler) } } } } - /// Builds the per-connection ``Connection`` and ``ConnectionContext`` for a HTTP/3 connection channel and - /// dispatches the connection to the connection handler. Errors from the connection handler are logged. + /// Builds the per-connection ``Connection`` for a HTTP/3 connection and dispatches it, alongside its + /// ``ConnectionContext``, to the connection handler. Errors from the connection handler are logged. func dispatchHTTP3Connection( - _ http3Connection: HTTP3ServerConnection, + _ inboundConnection: HTTP3InboundConnection, handler: Handler ) async { - let context = ConnectionContext( - httpVersion: .http3, - remoteAddress: nil, - localAddress: nil, - validatedPeerCertificateChain: nil - ) + let context = inboundConnection.context let connection = Connection( server: self, context: context, - httpProtocol: .http3(connection: http3Connection) + httpProtocol: .http3(connection: inboundConnection.connection) ) do { @@ -237,16 +241,16 @@ extension NIOHTTPServer { socket: socket, configuration: configuration, addressContinuation: addressContinuation - ) { _, multiplexer in + ) { _, inboundConnections in await self.serveHTTP3( - connectionMultiplexer: multiplexer, + inboundConnections: inboundConnections, connectionHandler: connectionHandler ) } } } - /// Provides a configured HTTP/3 channel and the associated connection multiplexer. The underlying socket is closed + /// Provides a configured HTTP/3 channel and the stream of connections it accepts. The underlying socket is closed /// when either returning or throwing from the `body` closure. func withHTTP3Channel( address: NIOCore.SocketAddress, @@ -254,8 +258,7 @@ extension NIOHTTPServer { socket: HTTP3ListenerSocket, configuration: ListenerConfiguration.HTTP3, addressContinuation: AsyncThrowingStream.Continuation, - _ body: (any Channel, HTTP3ServerConnectionMultiplexer) async throws -> - Void + _ body: (any Channel, AsyncStream) async throws -> Void ) async throws { var bootstrap = DatagramBootstrap(group: eventLoop) .channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1) @@ -265,17 +268,17 @@ extension NIOHTTPServer { } let quicChannel: any Channel - let multiplexer: HTTP3ServerConnectionMultiplexer + let inboundConnections: AsyncStream do { - (quicChannel, multiplexer) = try await bootstrap.bind(to: address) { channel in + (quicChannel, inboundConnections) = try await bootstrap.bind(to: address) { channel in channel.eventLoop.makeCompletedFuture { - let multiplexer = try self.setupQUICChannel( + let inboundConnections = try self.setupQUICChannel( channel: channel, configuration: configuration, socket: socket ) - return (channel, multiplexer) + return (channel, inboundConnections) } } } catch { @@ -318,7 +321,7 @@ extension NIOHTTPServer { addressContinuation.yield(localAddress) do { - try await body(quicChannel, multiplexer) + try await body(quicChannel, inboundConnections) } catch { try? await quicChannel.close() throw error @@ -335,8 +338,7 @@ extension NIOHTTPServer { } } - /// Installs the QUIC handler on a bound datagram channel and returns the channel alongside the connection - /// multiplexer. + /// Installs the QUIC handler on a bound datagram channel and returns the stream of connections it accepts. /// /// The QUIC connection channels `QUICHandler` creates are children of this datagram channel and run on its /// event loop, so a connection's whole lifetime stays on one loop. @@ -344,8 +346,13 @@ extension NIOHTTPServer { channel: any Channel, configuration: ListenerConfiguration.HTTP3, socket: HTTP3ListenerSocket - ) throws -> HTTP3ServerConnectionMultiplexer { - let connectionMultiplexer = HTTP3ServerConnectionMultiplexer() + ) throws -> AsyncStream { + // TODO: Go back to an `HTTP3ServerConnectionMultiplexer` once NIOHTTP3 lets each `HTTP3ServerConnection` carry + // per-connection state. Its multiplexer can carry only the connection, which doesn't expose its channel, so this stream + // carries the two together instead. + let (inboundConnections, inboundConnectionsContinuation) = AsyncStream.makeStream( + of: HTTP3InboundConnection.self + ) #if UnstableHTTPDatagrams let quicConfiguration = QUICConfiguration( @@ -376,12 +383,12 @@ extension NIOHTTPServer { logger: self.logger, inboundConnectionInitializer: { connectionChannel, streamCreator in connectionChannel.eventLoop.makeCompletedFuture { - let connection = try self.setupHTTP3Connection( + let inboundConnection = try self.setupHTTP3Connection( http3Configuration: configuration.http3Configuration, connectionChannel: connectionChannel, streamCreator: streamCreator ) - connectionMultiplexer.yield(connection: connection) + inboundConnectionsContinuation.yield(inboundConnection) } }, inboundStreamInitializer: { streamChannel in @@ -391,22 +398,34 @@ extension NIOHTTPServer { } }, noMoreConnections: { - connectionMultiplexer.finish() + inboundConnectionsContinuation.finish() }, connectionIDGenerator: connectionIDGenerator ) try channel.pipeline.syncOperations.addHandler(quicHandler) - return connectionMultiplexer + return inboundConnections } - /// Sets up an `HTTP3ConnectionHandler` and adds it to the connection channel pipeline. + /// Sets up an `HTTP3ConnectionHandler` and adds it to the connection channel pipeline, and builds the + /// connection's ``ConnectionContext``. func setupHTTP3Connection( http3Configuration: NIOHTTPServerConfiguration.HTTP3, connectionChannel: any Channel, streamCreator: NIOQUIC.QUICStreamCreator, - ) throws -> HTTP3ServerConnection { + ) throws -> HTTP3InboundConnection { + let context = ConnectionContext( + httpVersion: .http3, + remoteAddress: try? NIOHTTPServer.SocketAddress(connectionChannel.remoteAddress), + localAddress: try? NIOHTTPServer.SocketAddress(connectionChannel.localAddress), + validatedPeerCertificateChain: nil + ) + + // First in the pipeline, so that it sees the events the QUIC stack fires into the connection channel before + // any other handler can consume them. + try connectionChannel.pipeline.syncOperations.addHandler(ConnectionEventsHandler(connectionContext: context)) + let connectionEventLoop = connectionChannel.eventLoop let loopBoundHandler = NIOLoopBoundBox?>(nil, eventLoop: connectionEventLoop) @@ -506,7 +525,7 @@ extension NIOHTTPServer { try connectionChannel.pipeline.syncOperations.addHandlers([http3Handler, connectionManager]) #endif - return connection + return HTTP3InboundConnection(connection: connection, context: context) } /// Configures the pipeline for an inbound HTTP/3 stream channel and wraps it in a `NIOAsyncChannel`. diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+RequestContext.swift b/Sources/NIOHTTPServer/NIOHTTPServer+RequestContext.swift index 7d4f145..c77f548 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+RequestContext.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+RequestContext.swift @@ -23,6 +23,7 @@ extension NIOHTTPServer { /// Conforms to: /// - ``HTTPServerCapability/ConnectionInfo`` โ€” peer / local addresses. /// - ``HTTPServerCapability/PeerCertificate`` โ€” mTLS-validated peer chain. + /// - ``HTTPServerCapability/ConnectionEvents`` โ€” events on the connection carrying the request. /// /// Generic library code can constrain on these capabilities to access /// per-request data without depending on ``NIOHTTPServer`` directly. @@ -67,8 +68,16 @@ extension NIOHTTPServer.RequestContext: HTTPServerCapability.PeerCertificate { /// Returns `nil` when mTLS is not configured, or when the configured custom verification callback did not return /// the derived validated chain. public var validatedPeerCertificateChain: X509.ValidatedCertificateChain? { - get { - self.connectionContext.validatedPeerCertificateChain - } + self.connectionContext.validatedPeerCertificateChain + } +} + +@available(anyAppleOS 27.0, *) +extension NIOHTTPServer.RequestContext: HTTPServerCapability.ConnectionEvents { + /// The events that happen on the connection carrying this request, such as the peer's address changing. + /// + /// The sequence finishes once the connection closes. See ``NIOHTTPServer/ConnectionEvents``. + public var connectionEvents: NIOHTTPServer.ConnectionEvents { + self.connectionContext.connectionEvents } } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift index 95fbca7..673e381 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift @@ -39,7 +39,10 @@ extension NIOHTTPServer { } let channel: NegotiatedChannel - let validatedPeerCertificateChain: X509.ValidatedCertificateChain? + + /// The context built for the connection when its pipeline was set up, which the connection channel's + /// ``ConnectionEventsHandler`` keeps current. + let context: ConnectionContext } /// Serves incoming connections. Each connection undergoes ALPN negotiation to determine whether to use HTTP/1.1 or @@ -96,6 +99,8 @@ extension NIOHTTPServer { return } + let context = result.context + switch result.channel { case .http1_1(let requestChannel): // The dispatcher owns the channel's `executeThenClose` so the @@ -103,12 +108,6 @@ extension NIOHTTPServer { // connection handler called `handleRequests`. do { try await requestChannel.channel.executeThenClose { inbound, outbound in - let context = ConnectionContext( - httpVersion: .http1_1, - remoteAddress: try? NIOHTTPServer.SocketAddress(requestChannel.channel.channel.remoteAddress), - localAddress: try? NIOHTTPServer.SocketAddress(requestChannel.channel.channel.localAddress), - validatedPeerCertificateChain: result.validatedPeerCertificateChain - ) let connection = Connection( server: self, context: context, @@ -136,10 +135,6 @@ extension NIOHTTPServer { } case .http2(let connectionChannel, let multiplexer): - let context = NIOHTTPServer.makeHTTP2ConnectionContext( - connectionChannel: connectionChannel, - validatedPeerCertificateChain: result.validatedPeerCertificateChain - ) let connection = Connection( server: self, context: context, @@ -258,11 +253,16 @@ extension NIOHTTPServer { } } + /// Configures the HTTP/2 server pipeline, and builds the connection's ``ConnectionContext``. private func setupHTTP2Connection( channel: any Channel, configuration: NIOHTTPServerConfiguration.HTTP2 ) -> EventLoopFuture< - (any Channel, NIOHTTP2Handler.AsyncStreamMultiplexer) + ( + any Channel, + NIOHTTP2Handler.AsyncStreamMultiplexer, + ConnectionContext + ) > { channel.eventLoop.makeCompletedFuture { try channel.pipeline.syncOperations.configureAsyncHTTP2Pipeline( @@ -311,8 +311,16 @@ extension NIOHTTPServer { } ) } - .flatMap { multiplexer in - channel.eventLoop.makeCompletedFuture(.success((channel, multiplexer))) + .flatMapThrowing { multiplexer in + // This runs once ALPN has negotiated HTTP/2, by which point the handshake has validated any peer + // certificate chain. + let context = NIOHTTPServer.makeHTTP2ConnectionContext( + connectionChannel: channel, + validatedPeerCertificateChain: channel.extractPeerCertificateChain(logger: self.logger) + ) + try channel.pipeline.syncOperations.addHandler(ConnectionEventsHandler(connectionContext: context)) + + return (channel, multiplexer, context) } } @@ -343,11 +351,10 @@ extension NIOHTTPServer { return self.setupHTTP1_1Connection( channel: channel, isSecure: true - ).map { requestChannel in + ).map { connectionChannel in NegotiationResult( - channel: .http1_1(requestChannel), - validatedPeerCertificateChain: requestChannel.channel.channel - .extractPeerCertificateChain(logger: self.logger) + channel: .http1_1(connectionChannel.requestChannel), + context: connectionChannel.context ) } @@ -355,11 +362,8 @@ extension NIOHTTPServer { return self.setupHTTP2Connection( channel: channel, configuration: http2Config - ).map { (channel, streamMultiplexer) in - NegotiationResult( - channel: .http2(channel, streamMultiplexer), - validatedPeerCertificateChain: channel.extractPeerCertificateChain(logger: self.logger) - ) + ).map { (channel, streamMultiplexer, context) in + NegotiationResult(channel: .http2(channel, streamMultiplexer), context: context) } case (.negotiated, _), (.fallback, _): diff --git a/Sources/NIOHTTPServer/NIOHTTPServer.swift b/Sources/NIOHTTPServer/NIOHTTPServer.swift index 40eb530..529126d 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer.swift @@ -570,6 +570,17 @@ struct HTTPRequestChannelAndCancellationSignal: Sendable { var clientClosed: AsyncStream } +/// An HTTP/1.1 connection, whose channel is also the channel carrying its requests, and the context built for it when +/// its pipeline was set up. +/// +/// The context travels with the channel because the channel's ``ConnectionEventsHandler`` keeps that same context +/// current. +@available(anyAppleOS 27.0, *) +struct HTTP1ConnectionChannel: Sendable { + var requestChannel: HTTPRequestChannelAndCancellationSignal + var context: NIOHTTPServer.ConnectionContext +} + @available(anyAppleOS 27.0, *) extension ServerBootstrap { /// Makes a `ServerBootstrap` alongside the `ServerQuiescingHelper` used to later shut that listener down gracefully. diff --git a/Tests/NIOHTTPServerTests/ConnectionEventsTests.swift b/Tests/NIOHTTPServerTests/ConnectionEventsTests.swift new file mode 100644 index 0000000..9a7d359 --- /dev/null +++ b/Tests/NIOHTTPServerTests/ConnectionEventsTests.swift @@ -0,0 +1,382 @@ +//===----------------------------------------------------------------------===// +// +// 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 BasicContainers +import Logging +import NIOConcurrencyHelpers +import NIOCore +import NIOEmbedded +import Testing + +@testable import NIOHTTPServer + +@Suite("Connection events", .timeLimit(.minutes(1))) +struct ConnectionEventsTests { + static let logger = Logger(label: "ConnectionEventsTests") + + /// The address the client connected from, before any migration. + @available(anyAppleOS 27.0, *) + static var originalAddress: NIOHTTPServer.SocketAddress { .ipv4(host: "203.0.113.7", port: 50000) } + + @available(anyAppleOS 27.0, *) + static var migratedAddress: NIOHTTPServer.SocketAddress { .ipv4(host: "192.0.2.1", port: 4433) } + + @available(anyAppleOS 27.0, *) + static var secondMigratedAddress: NIOHTTPServer.SocketAddress { .ipv4(host: "192.0.2.2", port: 4434) } + + /// How long to wait for something that should happen promptly before concluding it never will. + static let deliveryWindow = Duration.seconds(5) + + // MARK: - The connection channel's handler + + @available(anyAppleOS 27.0, *) + @Test("An event fired into the connection's pipeline updates its remote address") + func eventUpdatesRemoteAddress() throws { + let context = NIOHTTPServer.ConnectionContext(httpVersion: .http2, remoteAddress: Self.originalAddress) + let channel = try Self.makeConnectionChannel(keepingCurrent: context) + + channel.pipeline.fireUserInboundEventTriggered( + NIOHTTPServer.ConnectionEvent.remoteAddressChanged(Self.migratedAddress) + ) + + #expect(context.remoteAddress == Self.migratedAddress) + } + + @available(anyAppleOS 27.0, *) + @Test("Every iterator receives each event that happens after it was created") + func everyIteratorReceivesEachEvent() async throws { + let context = NIOHTTPServer.ConnectionContext(httpVersion: .http2, remoteAddress: Self.originalAddress) + let channel = try Self.makeConnectionChannel(keepingCurrent: context) + var first = context.connectionEvents.makeAsyncIterator() + var second = context.connectionEvents.makeAsyncIterator() + + channel.pipeline.fireUserInboundEventTriggered( + NIOHTTPServer.ConnectionEvent.remoteAddressChanged(Self.migratedAddress) + ) + channel.pipeline.fireChannelInactive() + + #expect(await first.next() == .remoteAddressChanged(Self.migratedAddress)) + #expect(await second.next() == .remoteAddressChanged(Self.migratedAddress)) + } + + @available(anyAppleOS 27.0, *) + @Test("An iterator does not receive events that happened before it was created") + func iteratorSkipsEarlierEvents() async throws { + let context = NIOHTTPServer.ConnectionContext(httpVersion: .http2, remoteAddress: Self.originalAddress) + let channel = try Self.makeConnectionChannel(keepingCurrent: context) + + channel.pipeline.fireUserInboundEventTriggered( + NIOHTTPServer.ConnectionEvent.remoteAddressChanged(Self.migratedAddress) + ) + var iterator = context.connectionEvents.makeAsyncIterator() + channel.pipeline.fireUserInboundEventTriggered( + NIOHTTPServer.ConnectionEvent.remoteAddressChanged(Self.secondMigratedAddress) + ) + channel.pipeline.fireChannelInactive() + + #expect(await iterator.next() == .remoteAddressChanged(Self.secondMigratedAddress)) + } + + @available(anyAppleOS 27.0, *) + @Test("The connection going inactive ends every iterator, including ones created afterwards") + func inactiveConnectionEndsIterators() async throws { + let context = NIOHTTPServer.ConnectionContext(httpVersion: .http2, remoteAddress: Self.originalAddress) + let channel = try Self.makeConnectionChannel(keepingCurrent: context) + var before = context.connectionEvents.makeAsyncIterator() + + channel.pipeline.fireChannelInactive() + var after = context.connectionEvents.makeAsyncIterator() + + #expect(await before.next() == nil) + #expect(await after.next() == nil) + } + + @available(anyAppleOS 27.0, *) + @Test("Events carry on down the pipeline") + func eventsAreForwarded() throws { + let context = NIOHTTPServer.ConnectionContext(httpVersion: .http2, remoteAddress: Self.originalAddress) + let channel = try Self.makeConnectionChannel(keepingCurrent: context) + let recorder = UserEventRecorder() + try channel.pipeline.syncOperations.addHandler(recorder) + + channel.pipeline.fireUserInboundEventTriggered( + NIOHTTPServer.ConnectionEvent.remoteAddressChanged(Self.migratedAddress) + ) + channel.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed) + + #expect(recorder.events.count == 2) + #expect(recorder.events.first as? NIOHTTPServer.ConnectionEvent == .remoteAddressChanged(Self.migratedAddress)) + #expect(recorder.events.last as? ChannelEvent == .inputClosed) + } + + @available(anyAppleOS 27.0, *) + @Test("A dropped iterator is no longer delivered to") + func droppedIteratorIsUnregistered() throws { + let context = NIOHTTPServer.ConnectionContext(httpVersion: .http2, remoteAddress: Self.originalAddress) + + do { + let iterator = context.connectionEvents.makeAsyncIterator() + #expect(context.connectionEvents.broadcaster.subscriberCount == 1) + _ = consume iterator + } + + #expect(context.connectionEvents.broadcaster.subscriberCount == 0) + } + + // MARK: - End to end + + @available(anyAppleOS 27.0, *) + @Test( + "A request handler observes its connection's remote address changing", + arguments: NIOHTTPServer.HTTPVersion.allCases + ) + func requestHandlerObservesRemoteAddressChange(httpVersion: NIOHTTPServer.HTTPVersion) async throws { + let (server, clientConfiguration) = try TestHelpers.makeServerAndClientConfiguration( + for: httpVersion, + clientLogger: Self.logger, + serverLogger: Self.logger + ) + + let observed = NIOLockedValueBox< + (event: NIOHTTPServer.ConnectionEvent?, remoteAddress: NIOHTTPServer.SocketAddress?) + >((nil, nil)) + + try await TestHelpers.withClientServerConnection( + clientConfiguration: clientConfiguration, + server: server, + serverHandler: HTTPServerClosureRequestHandler { _, requestContext, _, responseSender in + // Stands in for the QUIC stack reporting a migration on the connection carrying this request. + let connectionChannel = Self.connectionChannel(carrying: requestContext) + let event = try await Self.firstEvent(of: requestContext.connectionEvents) { + connectionChannel.pipeline.fireUserInboundEventTriggered( + NIOHTTPServer.ConnectionEvent.remoteAddressChanged(Self.migratedAddress) + ) + } + observed.withLockedValue { $0 = (event, requestContext.remoteAddress) } + + var body = UniqueArray(copying: []) + try await responseSender.sendAndFinish(.init(status: .ok), buffer: &body) + } + ) { _, connection in + try await Self.sendRequestAndAwaitResponse(on: connection, httpVersion: httpVersion) + } + + let (event, remoteAddress) = observed.withLockedValue { $0 } + #expect(event == .remoteAddressChanged(Self.migratedAddress)) + #expect(remoteAddress == Self.migratedAddress) + } + + @available(anyAppleOS 27.0, *) + @Test( + "A connection's events finish when the client closes the connection", + arguments: NIOHTTPServer.HTTPVersion.allCases + ) + func connectionEventsFinishWhenClientCloses(httpVersion: NIOHTTPServer.HTTPVersion) async throws { + let (server, clientConfiguration) = try TestHelpers.makeServerAndClientConfiguration( + for: httpVersion, + clientLogger: Self.logger, + serverLogger: Self.logger + ) + + let (subscribed, subscribedContinuation) = AsyncStream.makeStream(of: Void.self) + let (finished, finishedContinuation) = AsyncStream.makeStream(of: Void.self) + + try await TestHelpers.withClientServerConnection( + clientConfiguration: clientConfiguration, + server: server, + connectionHandler: NIOHTTPServerClosureConnectionHandler { connection, context in + // Unstructured, so that observing the events cannot hold the connection open. + Task { + var iterator = context.connectionEvents.makeAsyncIterator() + subscribedContinuation.finish() + while await iterator.next() != nil {} + finishedContinuation.finish() + } + + await connection.handleRequests { _, _, _, _ in } + } + ) { _, connection in + for await _ in subscribed {} + + #if HTTP3 + if httpVersion == .http3 { + try await connection.closeAnnouncingDeparture() + } else { + try await connection.close() + } + #else + try await connection.close() + #endif + + #expect( + await Self.finishes(finished, within: Self.deliveryWindow), + "The connection's events should have finished once the client closed the connection (\(httpVersion))." + ) + } + } + + @available(anyAppleOS 27.0, *) + @Test( + "The request context reports the connection's local and remote addresses", + arguments: NIOHTTPServer.HTTPVersion.allCases + ) + func requestContextReportsAddresses(httpVersion: NIOHTTPServer.HTTPVersion) async throws { + let (server, clientConfiguration) = try TestHelpers.makeServerAndClientConfiguration( + for: httpVersion, + clientLogger: Self.logger, + serverLogger: Self.logger + ) + + let observed = NIOLockedValueBox< + (localAddress: NIOHTTPServer.SocketAddress?, remoteAddress: NIOHTTPServer.SocketAddress?) + >((nil, nil)) + + try await TestHelpers.withClientServerConnection( + clientConfiguration: clientConfiguration, + server: server, + serverHandler: HTTPServerClosureRequestHandler { _, requestContext, _, responseSender in + observed.withLockedValue { $0 = (requestContext.localAddress, requestContext.remoteAddress) } + + var body = UniqueArray(copying: []) + try await responseSender.sendAndFinish(.init(status: .ok), buffer: &body) + } + ) { serverAddress, connection in + // Read before the request: over HTTP/1.1 the request channel is the connection, which the request closes. + let clientPort = try #require(connection.localPort) + try await Self.sendRequestAndAwaitResponse(on: connection, httpVersion: httpVersion) + + let (localAddress, remoteAddress) = observed.withLockedValue { $0 } + #expect(localAddress?.port == serverAddress.port) + #expect(remoteAddress?.port == clientPort) + } + } + + // MARK: - Helpers + + /// An `EmbeddedChannel` standing in for a connection channel, whose pipeline keeps `context` current. + @available(anyAppleOS 27.0, *) + private static func makeConnectionChannel( + keepingCurrent context: NIOHTTPServer.ConnectionContext + ) throws -> EmbeddedChannel { + let channel = EmbeddedChannel() + try channel.pipeline.syncOperations.addHandler(ConnectionEventsHandler(connectionContext: context)) + return channel + } + + /// The channel of the connection carrying a request. + /// + /// Over HTTP/1.1 the request's channel is the connection channel itself, while over HTTP/2 and HTTP/3 it is a + /// stream channel whose parent is the connection channel. + @available(anyAppleOS 27.0, *) + private static func connectionChannel(carrying requestContext: NIOHTTPServer.RequestContext) -> any Channel { + switch requestContext.connectionContext.httpVersion { + case .plaintextHTTP1_1, .http1_1: + requestContext.channel + default: + requestContext.channel.parent! + } + } + + /// Starts iterating `events`, runs `trigger`, and returns the first event that follows, or `nil` if none arrives + /// within ``deliveryWindow``. + @available(anyAppleOS 27.0, *) + private static func firstEvent( + of events: NIOHTTPServer.ConnectionEvents, + after trigger: () async throws -> Void + ) async throws -> NIOHTTPServer.ConnectionEvent? { + let (subscribed, subscribedContinuation) = AsyncStream.makeStream(of: Void.self) + + return try await withThrowingTaskGroup(of: NIOHTTPServer.ConnectionEvent?.self) { group in + group.addTask { + var iterator = events.makeAsyncIterator() + subscribedContinuation.finish() + return await iterator.next() + } + group.addTask { + try? await Task.sleep(for: Self.deliveryWindow) + return nil + } + + for await _ in subscribed {} + try await trigger() + + let first = try await group.next() ?? nil + group.cancelAll() + return first + } + } + + /// Returns whether `stream` finishes within `window`. + private static func finishes(_ stream: AsyncStream, within window: Duration) async -> Bool { + await withTaskGroup(of: Bool.self) { group in + group.addTask { + for await _ in stream {} + return true + } + group.addTask { + try? await Task.sleep(for: window) + return false + } + + let first = await group.next() ?? false + group.cancelAll() + return first + } + } + + /// Sends a bodiless `GET` on a new request channel and reads the response through to its end. + @available(anyAppleOS 27.0, *) + private static func sendRequestAndAwaitResponse( + on connection: TestClientConnection, + httpVersion: NIOHTTPServer.HTTPVersion + ) async throws { + let requestChannel = try await connection.makeRequestChannel(expectedHTTPVersion: httpVersion) + try await requestChannel.executeThenClose { inbound, outbound in + try await outbound.write(.testHead(method: .get, for: httpVersion)) + try await outbound.write(.end(nil)) + + for try await part in inbound { + if case .end = part { break } + } + } + } +} + +/// Records the user inbound events that reach it. +private final class UserEventRecorder: ChannelInboundHandler { + typealias InboundIn = NIOAny + + private(set) var events: [Any] = [] + + func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) { + self.events.append(event) + } +} + +@available(anyAppleOS 27.0, *) +extension TestClientConnection { + /// The port this client connection is bound to locally. + var localPort: Int? { + switch self.connectionProtocol { + case .http1(let connectionChannel): + connectionChannel.channel.localAddress?.port + case .http2(let connectionChannel, _): + connectionChannel.localAddress?.port + #if HTTP3 + case .http3(_, let connectionChannel, _): + connectionChannel.localAddress?.port + #endif + } + } +} diff --git a/Tests/NIOHTTPServerTests/NIOHTTPServer+ServiceLifecycleTests.swift b/Tests/NIOHTTPServerTests/NIOHTTPServer+ServiceLifecycleTests.swift index c56227a..9501bb1 100644 --- a/Tests/NIOHTTPServerTests/NIOHTTPServer+ServiceLifecycleTests.swift +++ b/Tests/NIOHTTPServerTests/NIOHTTPServer+ServiceLifecycleTests.swift @@ -41,6 +41,7 @@ struct NIOHTTPServiceLifecycleTests { @Test( "Active connection completes when graceful shutdown triggered", + .timeLimit(.minutes(1)), arguments: [NIOHTTPServer.HTTPVersion.http1_1, .http2] ) @available(anyAppleOS 27.0, *) diff --git a/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1.swift b/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1.swift index 8253e44..2958f9b 100644 --- a/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1.swift +++ b/Tests/NIOHTTPServerTests/Utilities/TestingChannelClientServer/NIOHTTPServer+HTTP1.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( + try NIOAsyncChannel( wrappingChannelSynchronously: testChannel, configuration: .init() ) From f9d623250010a91c6cf0f5e145582b6e77c4a20f Mon Sep 17 00:00:00 2001 From: Gus Cairo Date: Mon, 12 Oct 2026 09:26:07 +0100 Subject: [PATCH 2/2] Remove handler from h1/h2 pipelines --- .../ConnectionEventsHandler.swift | 3 ++ .../NIOHTTPServer+ConnectionContext.swift | 4 +-- .../NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift | 6 +++- .../NIOHTTPServer+SecureUpgrade.swift | 13 +++++--- Sources/NIOHTTPServer/NIOHTTPServer.swift | 4 +-- .../ConnectionEventsTests.swift | 33 ++++++------------- 6 files changed, 31 insertions(+), 32 deletions(-) diff --git a/Sources/NIOHTTPServer/ConnectionEventsHandler.swift b/Sources/NIOHTTPServer/ConnectionEventsHandler.swift index 582f9fa..c46ce83 100644 --- a/Sources/NIOHTTPServer/ConnectionEventsHandler.swift +++ b/Sources/NIOHTTPServer/ConnectionEventsHandler.swift @@ -17,6 +17,9 @@ import NIOCore /// This channel handler keeps a connection's ``NIOHTTPServer/ConnectionContext`` current with the events fired into the /// connection channel's pipeline, and reports them through the context's /// ``NIOHTTPServer/ConnectionContext/connectionEvents``, which it finishes once the connection closes. +/// +/// Only HTTP/3 connection channels have one: a QUIC connection can migrate to a new network path, whereas a TCP +/// connection's path is fixed for its lifetime. @available(anyAppleOS 27.0, *) final class ConnectionEventsHandler: ChannelInboundHandler, RemovableChannelHandler { typealias InboundIn = NIOAny diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift b/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift index b5bef8f..8a3cc60 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+ConnectionContext.swift @@ -70,8 +70,8 @@ extension NIOHTTPServer { /// The sequence finishes once the connection closes. See ``ConnectionEvents``. public let connectionEvents: ConnectionEvents - /// Shared by every copy of this context, so that the connection's ``ConnectionEventsHandler`` keeps them all - /// current. + /// Shared by every copy of this context, so that over HTTP/3 the connection's ``ConnectionEventsHandler`` keeps + /// them all current. private let currentRemoteAddress: NIOLockedValueBox init( diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift index 1d18cc0..2351809 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+HTTP1_1.swift @@ -163,7 +163,11 @@ extension NIOHTTPServer { ClientClosedMonitor(clientClosed: clientClosedContinuation) ) - try channel.pipeline.syncOperations.addHandler(ConnectionEventsHandler(connectionContext: context)) + // A TCP connection cannot migrate to a new network path, so no event ever changes this context: all that + // is left is to end its events once the connection closes. + channel.closeFuture.whenComplete { _ in + context.unsubscribeFromConnectionEvents() + } return HTTP1ConnectionChannel( requestChannel: HTTPRequestChannelAndCancellationSignal( diff --git a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift index 673e381..181c356 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer+SecureUpgrade.swift @@ -40,8 +40,8 @@ extension NIOHTTPServer { let channel: NegotiatedChannel - /// The context built for the connection when its pipeline was set up, which the connection channel's - /// ``ConnectionEventsHandler`` keeps current. + /// The context built for the connection when its pipeline was set up, whose + /// ``ConnectionContext/connectionEvents`` end when the connection channel closes. let context: ConnectionContext } @@ -311,14 +311,19 @@ extension NIOHTTPServer { } ) } - .flatMapThrowing { multiplexer in + .map { multiplexer in // This runs once ALPN has negotiated HTTP/2, by which point the handshake has validated any peer // certificate chain. let context = NIOHTTPServer.makeHTTP2ConnectionContext( connectionChannel: channel, validatedPeerCertificateChain: channel.extractPeerCertificateChain(logger: self.logger) ) - try channel.pipeline.syncOperations.addHandler(ConnectionEventsHandler(connectionContext: context)) + + // A TCP connection cannot migrate to a new network path, so no event ever changes this context: all that + // is left is to end its events once the connection closes. + channel.closeFuture.whenComplete { _ in + context.unsubscribeFromConnectionEvents() + } return (channel, multiplexer, context) } diff --git a/Sources/NIOHTTPServer/NIOHTTPServer.swift b/Sources/NIOHTTPServer/NIOHTTPServer.swift index 529126d..59d8ca2 100644 --- a/Sources/NIOHTTPServer/NIOHTTPServer.swift +++ b/Sources/NIOHTTPServer/NIOHTTPServer.swift @@ -573,8 +573,8 @@ struct HTTPRequestChannelAndCancellationSignal: Sendable { /// An HTTP/1.1 connection, whose channel is also the channel carrying its requests, and the context built for it when /// its pipeline was set up. /// -/// The context travels with the channel because the channel's ``ConnectionEventsHandler`` keeps that same context -/// current. +/// The context travels with the channel because the channel closing ends that same context's +/// ``NIOHTTPServer/ConnectionContext/connectionEvents``. @available(anyAppleOS 27.0, *) struct HTTP1ConnectionChannel: Sendable { var requestChannel: HTTPRequestChannelAndCancellationSignal diff --git a/Tests/NIOHTTPServerTests/ConnectionEventsTests.swift b/Tests/NIOHTTPServerTests/ConnectionEventsTests.swift index 9a7d359..0314701 100644 --- a/Tests/NIOHTTPServerTests/ConnectionEventsTests.swift +++ b/Tests/NIOHTTPServerTests/ConnectionEventsTests.swift @@ -136,14 +136,13 @@ struct ConnectionEventsTests { // MARK: - End to end + #if HTTP3 + /// Only HTTP/3 connections can migrate to a new network path, so only their pipelines report connection events. @available(anyAppleOS 27.0, *) - @Test( - "A request handler observes its connection's remote address changing", - arguments: NIOHTTPServer.HTTPVersion.allCases - ) - func requestHandlerObservesRemoteAddressChange(httpVersion: NIOHTTPServer.HTTPVersion) async throws { + @Test("A request handler observes its connection's remote address changing") + func requestHandlerObservesRemoteAddressChange() async throws { let (server, clientConfiguration) = try TestHelpers.makeServerAndClientConfiguration( - for: httpVersion, + for: .http3, clientLogger: Self.logger, serverLogger: Self.logger ) @@ -156,8 +155,9 @@ struct ConnectionEventsTests { clientConfiguration: clientConfiguration, server: server, serverHandler: HTTPServerClosureRequestHandler { _, requestContext, _, responseSender in - // Stands in for the QUIC stack reporting a migration on the connection carrying this request. - let connectionChannel = Self.connectionChannel(carrying: requestContext) + // Stands in for the QUIC stack reporting a migration on the connection carrying this request. The + // request's stream channel is a child of that connection's channel. + let connectionChannel = requestContext.channel.parent! let event = try await Self.firstEvent(of: requestContext.connectionEvents) { connectionChannel.pipeline.fireUserInboundEventTriggered( NIOHTTPServer.ConnectionEvent.remoteAddressChanged(Self.migratedAddress) @@ -169,13 +169,14 @@ struct ConnectionEventsTests { try await responseSender.sendAndFinish(.init(status: .ok), buffer: &body) } ) { _, connection in - try await Self.sendRequestAndAwaitResponse(on: connection, httpVersion: httpVersion) + try await Self.sendRequestAndAwaitResponse(on: connection, httpVersion: .http3) } let (event, remoteAddress) = observed.withLockedValue { $0 } #expect(event == .remoteAddressChanged(Self.migratedAddress)) #expect(remoteAddress == Self.migratedAddress) } + #endif @available(anyAppleOS 27.0, *) @Test( @@ -274,20 +275,6 @@ struct ConnectionEventsTests { return channel } - /// The channel of the connection carrying a request. - /// - /// Over HTTP/1.1 the request's channel is the connection channel itself, while over HTTP/2 and HTTP/3 it is a - /// stream channel whose parent is the connection channel. - @available(anyAppleOS 27.0, *) - private static func connectionChannel(carrying requestContext: NIOHTTPServer.RequestContext) -> any Channel { - switch requestContext.connectionContext.httpVersion { - case .plaintextHTTP1_1, .http1_1: - requestContext.channel - default: - requestContext.channel.parent! - } - } - /// Starts iterating `events`, runs `trigger`, and returns the first event that follows, or `nil` if none arrives /// within ``deliveryWindow``. @available(anyAppleOS 27.0, *)