Enable strict concurrency checking for NIOHTTP1 (#3115)

### Motivation:

To ensure NIOHTTP1 concurrency safety.

### Modifications:

* Enable strict concurrency checking in the package manifest.
* Mark several objects `Sendable` with `@preconcurrency` annotations
where they are returned in futures which may execute in arbitrary
concurrency domains.
  * `NIOTypedHTTPClientProtocolUpgrader`
  * `NIOTypedHTTPClientUpgradeConfiguration`
  * `NIOUpgradableHTTPServerPipelineConfiguration`
  * `NIOUpgradableHTTPClientPipelineConfiguration`
  * `NIOTypedHTTPServerProtocolUpgrader`
  * `NIOTypedHTTPServerUpgradeConfiguration`
* Mark handlers as explicitly not sendable
  * `NIOTypedHTTPClientUpgradeHandler`
  * `NIOTypedHTTPServerUpgradeHandler`
* Added new Sendable type aliases:
  * `NIOHTTPClientUpgradeSendableConfiguration`
  * `NIOHTTPServerUpgradeSendableConfiguration`

### Result:

No more concurrency warnings. Builds will warn and CI will fail if
regressions are introduced.

Author: rnro

Package.swift

@@ -196,7 +196,8 @@ let package = Package(
                 "NIOConcurrencyHelpers",
                 "CNIOLLHTTP",
                 swiftCollections,
-            ]
+            ],
+            swiftSettings: strictConcurrencySettings
         ),
         .target(
             name: "NIOWebSocket",

Sources/NIOHTTP1/HTTPPipelineSetup.swift

@@ -22,6 +22,10 @@ public typealias NIOHTTPClientUpgradeConfiguration = (
     upgraders: [NIOHTTPClientProtocolUpgrader], completionHandler: @Sendable (ChannelHandlerContext) -> Void
 )
 
+public typealias NIOHTTPClientUpgradeSendableConfiguration = (
+    upgraders: [NIOHTTPClientProtocolUpgrader & Sendable], completionHandler: @Sendable (ChannelHandlerContext) -> Void
+)
+
 /// Configuration required to configure a HTTP server pipeline for upgrade.
 ///
 /// See the documentation for `HTTPServerUpgradeHandler` for details on these
@@ -33,6 +37,10 @@ public typealias NIOHTTPServerUpgradeConfiguration = (
     upgraders: [HTTPServerProtocolUpgrader], completionHandler: @Sendable (ChannelHandlerContext) -> Void
 )
 
+public typealias NIOHTTPServerUpgradeSendableConfiguration = (
+    upgraders: [HTTPServerProtocolUpgrader & Sendable], completionHandler: @Sendable (ChannelHandlerContext) -> Void
+)
+
 extension ChannelPipeline {
     /// Configure a `ChannelPipeline` for use as a HTTP client.
     ///
@@ -67,7 +75,7 @@ extension ChannelPipeline {
     public func addHTTPClientHandlers(
         position: Position = .last,
         leftOverBytesStrategy: RemoveAfterUpgradeStrategy = .dropBytes,
-        withClientUpgrade upgrade: NIOHTTPClientUpgradeConfiguration?
+        withClientUpgrade upgrade: NIOHTTPClientUpgradeSendableConfiguration?
     ) -> EventLoopFuture<Void> {
         self._addHTTPClientHandlers(
             position: position,
@@ -79,7 +87,7 @@ extension ChannelPipeline {
     private func _addHTTPClientHandlers(
         position: Position = .last,
         leftOverBytesStrategy: RemoveAfterUpgradeStrategy = .dropBytes,
-        withClientUpgrade upgrade: NIOHTTPClientUpgradeConfiguration?
+        withClientUpgrade upgrade: NIOHTTPClientUpgradeSendableConfiguration?
     ) -> EventLoopFuture<Void> {
         let future: EventLoopFuture<Void>
 
@@ -120,11 +128,12 @@ extension ChannelPipeline {
     ///         the upgrade completion handler. See the documentation on ``NIOHTTPClientUpgradeHandler``
     ///         for more details.
     /// - Returns: An `EventLoopFuture` that will fire when the pipeline is configured.
+    @preconcurrency
     public func addHTTPClientHandlers(
         position: Position = .last,
         leftOverBytesStrategy: RemoveAfterUpgradeStrategy = .dropBytes,
         enableOutboundHeaderValidation: Bool = true,
-        withClientUpgrade upgrade: NIOHTTPClientUpgradeConfiguration? = nil
+        withClientUpgrade upgrade: NIOHTTPClientUpgradeSendableConfiguration? = nil
     ) -> EventLoopFuture<Void> {
         let future: EventLoopFuture<Void>
 
@@ -168,12 +177,13 @@ extension ChannelPipeline {
     ///         the upgrade completion handler. See the documentation on ``NIOHTTPClientUpgradeHandler``
     ///         for more details.
     /// - Returns: An `EventLoopFuture` that will fire when the pipeline is configured.
+    @preconcurrency
     public func addHTTPClientHandlers(
         position: Position = .last,
         leftOverBytesStrategy: RemoveAfterUpgradeStrategy = .dropBytes,
         enableOutboundHeaderValidation: Bool = true,
         encoderConfiguration: HTTPRequestEncoder.Configuration = .init(),
-        withClientUpgrade upgrade: NIOHTTPClientUpgradeConfiguration? = nil
+        withClientUpgrade upgrade: NIOHTTPClientUpgradeSendableConfiguration? = nil
     ) -> EventLoopFuture<Void> {
         let future: EventLoopFuture<Void>
 
@@ -234,7 +244,7 @@ extension ChannelPipeline {
     public func configureHTTPServerPipeline(
         position: ChannelPipeline.Position = .last,
         withPipeliningAssistance pipelining: Bool = true,
-        withServerUpgrade upgrade: NIOHTTPServerUpgradeConfiguration? = nil,
+        withServerUpgrade upgrade: NIOHTTPServerUpgradeSendableConfiguration? = nil,
         withErrorHandling errorHandling: Bool = true
     ) -> EventLoopFuture<Void> {
         self._configureHTTPServerPipeline(
@@ -274,10 +284,11 @@ extension ChannelPipeline {
     ///   - headerValidation: Whether to validate outbound request headers to confirm that they meet
     ///         spec compliance. Defaults to `true`.
     /// - Returns: An `EventLoopFuture` that will fire when the pipeline is configured.
+    @preconcurrency
     public func configureHTTPServerPipeline(
         position: ChannelPipeline.Position = .last,
         withPipeliningAssistance pipelining: Bool = true,
-        withServerUpgrade upgrade: NIOHTTPServerUpgradeConfiguration? = nil,
+        withServerUpgrade upgrade: NIOHTTPServerUpgradeSendableConfiguration? = nil,
         withErrorHandling errorHandling: Bool = true,
         withOutboundHeaderValidation headerValidation: Bool = true
     ) -> EventLoopFuture<Void> {
@@ -320,10 +331,11 @@ extension ChannelPipeline {
     ///         spec compliance. Defaults to `true`.
     ///   - encoderConfiguration: The configuration for the ``HTTPResponseEncoder``.
     /// - Returns: An `EventLoopFuture` that will fire when the pipeline is configured.
+    @preconcurrency
     public func configureHTTPServerPipeline(
         position: ChannelPipeline.Position = .last,
         withPipeliningAssistance pipelining: Bool = true,
-        withServerUpgrade upgrade: NIOHTTPServerUpgradeConfiguration? = nil,
+        withServerUpgrade upgrade: NIOHTTPServerUpgradeSendableConfiguration? = nil,
         withErrorHandling errorHandling: Bool = true,
         withOutboundHeaderValidation headerValidation: Bool = true,
         withEncoderConfiguration encoderConfiguration: HTTPResponseEncoder.Configuration = .init()
@@ -341,7 +353,7 @@ extension ChannelPipeline {
     private func _configureHTTPServerPipeline(
         position: ChannelPipeline.Position = .last,
         withPipeliningAssistance pipelining: Bool = true,
-        withServerUpgrade upgrade: NIOHTTPServerUpgradeConfiguration? = nil,
+        withServerUpgrade upgrade: NIOHTTPServerUpgradeSendableConfiguration? = nil,
         withErrorHandling errorHandling: Bool = true,
         withOutboundHeaderValidation headerValidation: Bool = true,
         withEncoderConfiguration encoderConfiguration: HTTPResponseEncoder.Configuration = .init()

Sources/NIOHTTP1/HTTPServerUpgradeHandler.swift

@@ -180,28 +180,26 @@ public final class HTTPServerUpgradeHandler: ChannelInboundHandler, RemovableCha
         // We'll attempt to upgrade. This may take a while, so while we're waiting more data can come in.
         self.upgradeState = .awaitingUpgrader
 
-        let eventLoop = context.eventLoop
-        let loopBoundContext = context.loopBound
         self.handleUpgrade(context: context, request: request, requestedProtocols: requestedProtocols)
-            .hop(to: eventLoop)  // the user might return a future from another EventLoop.
             .whenSuccess { callback in
-                eventLoop.assertInEventLoop()
                 if let callback = callback {
                     self.gotUpgrader(upgrader: callback)
                 } else {
-                    self.notUpgrading(context: loopBoundContext.value, data: requestPart)
+                    self.notUpgrading(context: context, data: requestPart)
                 }
             }
     }
 
     /// The core of the upgrade handling logic.
     ///
-    /// - Returns: An `EventLoopFuture` that will contain a callback to invoke if upgrade is requested, or nil if upgrade has failed. Never returns a failed future.
+    /// - Returns: An isolated `EventLoopFuture` that will contain a callback to invoke if upgrade is requested,
+    /// or nil if upgrade has failed. Never returns a failed future.
     private func handleUpgrade(
         context: ChannelHandlerContext,
         request: HTTPRequestHead,
         requestedProtocols: [String]
-    ) -> EventLoopFuture<(() -> Void)?> {
+    ) -> EventLoopFuture<(() -> Void)?>.Isolated {
+
         let connectionHeader = Set(request.headers[canonicalForm: "connection"].map { $0.lowercased() })
         let allHeaderNames = Set(request.headers.map { $0.name.lowercased() })
 
@@ -219,18 +217,21 @@ public final class HTTPServerUpgradeHandler: ChannelInboundHandler, RemovableCha
     /// Attempt to upgrade a single protocol.
     ///
     /// Will recurse through `protocolIterator` if upgrade fails.
+    ///
+    /// - Returns: An isolated `EventLoopFuture` that will contain a callback to invoke if upgrade is requested,
+    /// or nil if upgrade has failed. Never returns a failed future.
     private func handleUpgradeForProtocol(
         context: ChannelHandlerContext,
         protocolIterator: Array<String>.Iterator,
         request: HTTPRequestHead,
         allHeaderNames: Set<String>,
         connectionHeader: Set<String>
-    ) -> EventLoopFuture<(() -> Void)?> {
+    ) -> EventLoopFuture<(() -> Void)?>.Isolated {
         // We want a local copy of the protocol iterator. We'll pass it to the next invocation of the function.
         var protocolIterator = protocolIterator
         guard let proto = protocolIterator.next() else {
             // We're done! No suitable protocol for upgrade.
-            return context.eventLoop.makeSucceededFuture(nil)
+            return context.eventLoop.makeSucceededIsolatedFuture(nil)
         }
 
         guard let upgrader = self.upgraders[proto.lowercased()] else {
@@ -256,66 +257,67 @@ public final class HTTPServerUpgradeHandler: ChannelInboundHandler, RemovableCha
 
         let responseHeaders = self.buildUpgradeHeaders(protocol: proto)
         let pipeline = context.pipeline
-        let loopBoundContext = context.loopBound
+
         return upgrader.buildUpgradeResponse(
             channel: context.channel,
             upgradeRequest: request,
             initialResponseHeaders: responseHeaders
-        ).map { finalResponseHeaders in
-            {
-                // Ok, we're upgrading.
-                self.upgradeState = .upgrading
-
-                // Before we finish the upgrade we have to remove the HTTPDecoder and any other non-Encoder HTTP
-                // handlers from the pipeline, to prevent them parsing any more data. We'll buffer the data until
-                // that completes.
-                // While there are a lot of Futures involved here it's quite possible that all of this code will
-                // actually complete synchronously: we just want to program for the possibility that it won't.
-                // Once that's done, we send the upgrade response, then remove the HTTP encoder, then call the
-                // internal handler, then call the user code, and then finally when the user code is done we do
-                // our final cleanup steps, namely we replay the received data we buffered in the meantime and
-                // then remove ourselves from the pipeline.
-                self.removeExtraHandlers(pipeline: pipeline).flatMap {
-                    self.sendUpgradeResponse(
-                        context: loopBoundContext.value,
-                        upgradeRequest: request,
-                        responseHeaders: finalResponseHeaders
-                    )
-                }.flatMap {
-                    pipeline.syncOperations.removeHandler(self.httpEncoder)
-                }.flatMap { () -> EventLoopFuture<Void> in
-                    let context = loopBoundContext.value
-                    self.upgradeCompletionHandler(context)
-                    return upgrader.upgrade(context: context, upgradeRequest: request)
-                }.whenComplete { result in
-                    let context = loopBoundContext.value
-                    switch result {
-                    case .success:
-                        context.fireUserInboundEventTriggered(
-                            HTTPServerUpgradeEvents.upgradeComplete(toProtocol: proto, upgradeRequest: request)
-                        )
-                        self.upgradeState = .upgradeComplete
-                        // When we remove ourselves we'll be delivering any buffered data.
-                        context.pipeline.syncOperations.removeHandler(context: context, promise: nil)
-
-                    case .failure(let error):
-                        // Remain in the '.upgrading' state.
-                        context.fireErrorCaught(error)
-                    }
+        ).hop(to: context.eventLoop)
+            .assumeIsolated()
+            .map { finalResponseHeaders in
+                {
+                    // Ok, we're upgrading.
+                    self.upgradeState = .upgrading
+
+                    // Before we finish the upgrade we have to remove the HTTPDecoder and any other non-Encoder HTTP
+                    // handlers from the pipeline, to prevent them parsing any more data. We'll buffer the data until
+                    // that completes.
+                    // While there are a lot of Futures involved here it's quite possible that all of this code will
+                    // actually complete synchronously: we just want to program for the possibility that it won't.
+                    // Once that's done, we send the upgrade response, then remove the HTTP encoder, then call the
+                    // internal handler, then call the user code, and then finally when the user code is done we do
+                    // our final cleanup steps, namely we replay the received data we buffered in the meantime and
+                    // then remove ourselves from the pipeline.
+                    self.removeExtraHandlers(pipeline: pipeline)
+                        .assumeIsolated()
+                        .flatMap {
+                            self.sendUpgradeResponse(
+                                context: context,
+                                upgradeRequest: request,
+                                responseHeaders: finalResponseHeaders
+                            )
+                        }.flatMap {
+                            pipeline.syncOperations.removeHandler(self.httpEncoder)
+                        }.flatMap { () -> EventLoopFuture<Void> in
+                            self.upgradeCompletionHandler(context)
+                            return upgrader.upgrade(context: context, upgradeRequest: request)
+                        }.whenComplete { result in
+                            switch result {
+                            case .success:
+                                context.fireUserInboundEventTriggered(
+                                    HTTPServerUpgradeEvents.upgradeComplete(toProtocol: proto, upgradeRequest: request)
+                                )
+                                self.upgradeState = .upgradeComplete
+                                // When we remove ourselves we'll be delivering any buffered data.
+                                context.pipeline.syncOperations.removeHandler(context: context, promise: nil)
+
+                            case .failure(let error):
+                                // Remain in the '.upgrading' state.
+                                context.fireErrorCaught(error)
+                            }
+                        }
                 }
+            }.flatMapError { error in
+                // No upgrade here. We want to fire the error down the pipeline, and then try another loop iteration.
+                context.fireErrorCaught(error)
+                return self.handleUpgradeForProtocol(
+                    context: context,
+                    protocolIterator: protocolIterator,
+                    request: request,
+                    allHeaderNames: allHeaderNames,
+                    connectionHeader: connectionHeader
+                )
             }
-        }.flatMapError { error in
-            // No upgrade here. We want to fire the error down the pipeline, and then try another loop iteration.
-            let context = loopBoundContext.value
-            context.fireErrorCaught(error)
-            return self.handleUpgradeForProtocol(
-                context: context,
-                protocolIterator: protocolIterator,
-                request: request,
-                allHeaderNames: allHeaderNames,
-                connectionHeader: connectionHeader
-            )
-        }
     }
 
     private func gotUpgrader(upgrader: @escaping (() -> Void)) {
@@ -379,7 +381,7 @@ public final class HTTPServerUpgradeHandler: ChannelInboundHandler, RemovableCha
         }
 
         return .andAllSucceed(
-            self.extraHTTPHandlers.map { pipeline.removeHandler($0) },
+            self.extraHTTPHandlers.map { pipeline.syncOperations.removeHandler($0) },
             on: pipeline.eventLoop
         )
     }

Sources/NIOHTTP1/HTTPTypedPipelineSetup.swift

@@ -18,7 +18,7 @@ import NIOCore
 
 /// Configuration for an upgradable HTTP pipeline.
 @available(macOS 13, iOS 16, tvOS 16, watchOS 9, *)
-public struct NIOUpgradableHTTPServerPipelineConfiguration<UpgradeResult: Sendable> {
+public struct NIOUpgradableHTTPServerPipelineConfiguration<UpgradeResult: Sendable>: Sendable {
     /// Whether to provide assistance handling HTTP clients that pipeline
     /// their requests. Defaults to `true`. If `false`, users will need to handle clients that pipeline themselves.
     public var enablePipelining = true
@@ -146,7 +146,7 @@ extension ChannelPipeline.SynchronousOperations {
 
 /// Configuration for an upgradable HTTP pipeline.
 @available(macOS 13, iOS 16, tvOS 16, watchOS 9, *)
-public struct NIOUpgradableHTTPClientPipelineConfiguration<UpgradeResult: Sendable> {
+public struct NIOUpgradableHTTPClientPipelineConfiguration<UpgradeResult: Sendable>: Sendable {
     /// The strategy to use when dealing with leftover bytes after removing the ``HTTPDecoder`` from the pipeline.
     public var leftOverBytesStrategy = RemoveAfterUpgradeStrategy.dropBytes