Improve errors and testing using NIOTS

Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool+Factory.swift

@@ -47,6 +47,7 @@ protocol HTTPConnectionRequester {
     func http1ConnectionCreated(_: HTTP1Connection)
     func http2ConnectionCreated(_: HTTP2Connection, maximumStreams: Int)
     func failedToCreateHTTPConnection(_: HTTPConnectionPool.Connection.ID, error: Error)
+    func waitingForConnectivity(_: HTTPConnectionPool.Connection.ID, error: Error)
 }
 
 extension HTTPConnectionPool.ConnectionFactory {
@@ -62,7 +63,7 @@ extension HTTPConnectionPool.ConnectionFactory {
         var logger = logger
         logger[metadataKey: "ahc-connection-id"] = "\(connectionID)"
 
-        self.makeChannel(connectionID: connectionID, deadline: deadline, eventLoop: eventLoop, logger: logger).whenComplete { result in
+        self.makeChannel(requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop, logger: logger).whenComplete { result in
             switch result {
             case .success(.http1_1(let channel)):
                 do {
@@ -104,13 +105,15 @@ extension HTTPConnectionPool.ConnectionFactory {
         case http2(Channel)
     }
 
-    func makeHTTP1Channel(
+    func makeHTTP1Channel<Requester: HTTPConnectionRequester>(
+        requester: Requester,
         connectionID: HTTPConnectionPool.Connection.ID,
         deadline: NIODeadline,
         eventLoop: EventLoop,
         logger: Logger
     ) -> EventLoopFuture<Channel> {
         self.makeChannel(
+            requester: requester,
             connectionID: connectionID,
             deadline: deadline,
             eventLoop: eventLoop,
@@ -137,7 +140,8 @@ extension HTTPConnectionPool.ConnectionFactory {
         }
     }
 
-    func makeChannel(
+    func makeChannel<Requester: HTTPConnectionRequester>(
+        requester: Requester,
         connectionID: HTTPConnectionPool.Connection.ID,
         deadline: NIODeadline,
         eventLoop: EventLoop,
@@ -150,6 +154,7 @@ extension HTTPConnectionPool.ConnectionFactory {
             case .socks:
                 channelFuture = self.makeSOCKSProxyChannel(
                     proxy,
+                    requester: requester,
                     connectionID: connectionID,
                     deadline: deadline,
                     eventLoop: eventLoop,
@@ -158,14 +163,15 @@ extension HTTPConnectionPool.ConnectionFactory {
             case .http:
                 channelFuture = self.makeHTTPProxyChannel(
                     proxy,
+                    requester: requester,
                     connectionID: connectionID,
                     deadline: deadline,
                     eventLoop: eventLoop,
                     logger: logger
                 )
             }
         } else {
-            channelFuture = self.makeNonProxiedChannel(deadline: deadline, eventLoop: eventLoop, logger: logger)
+            channelFuture = self.makeNonProxiedChannel(requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop, logger: logger)
         }
 
         // let's map `ChannelError.connectTimeout` into a `HTTPClientError.connectTimeout`
@@ -179,30 +185,38 @@ extension HTTPConnectionPool.ConnectionFactory {
         }
     }
 
-    private func makeNonProxiedChannel(
+    private func makeNonProxiedChannel<Requester: HTTPConnectionRequester>(
+        requester: Requester,
+        connectionID: HTTPConnectionPool.Connection.ID,
         deadline: NIODeadline,
         eventLoop: EventLoop,
         logger: Logger
     ) -> EventLoopFuture<NegotiatedProtocol> {
         switch self.key.scheme {
         case .http, .httpUnix, .unix:
-            return self.makePlainChannel(deadline: deadline, eventLoop: eventLoop).map { .http1_1($0) }
+            return self.makePlainChannel(requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop).map { .http1_1($0) }
         case .https, .httpsUnix:
-            return self.makeTLSChannel(deadline: deadline, eventLoop: eventLoop, logger: logger).flatMapThrowing {
+            return self.makeTLSChannel(requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop, logger: logger).flatMapThrowing {
                 channel, negotiated in
 
                 try self.matchALPNToHTTPVersion(negotiated, channel: channel)
             }
         }
     }
 
-    private func makePlainChannel(deadline: NIODeadline, eventLoop: EventLoop) -> EventLoopFuture<Channel> {
+    private func makePlainChannel<Requester: HTTPConnectionRequester>(
+        requester: Requester,
+        connectionID: HTTPConnectionPool.Connection.ID,
+        deadline: NIODeadline,
+        eventLoop: EventLoop
+    ) -> EventLoopFuture<Channel> {
         precondition(!self.key.scheme.usesTLS, "Unexpected scheme")
-        return self.makePlainBootstrap(deadline: deadline, eventLoop: eventLoop).connect(target: self.key.connectionTarget)
+        return self.makePlainBootstrap(requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop).connect(target: self.key.connectionTarget)
     }
 
-    private func makeHTTPProxyChannel(
+    private func makeHTTPProxyChannel<Requester: HTTPConnectionRequester>(
         _ proxy: HTTPClient.Configuration.Proxy,
+        requester: Requester,
         connectionID: HTTPConnectionPool.Connection.ID,
         deadline: NIODeadline,
         eventLoop: EventLoop,
@@ -211,7 +225,7 @@ extension HTTPConnectionPool.ConnectionFactory {
         // A proxy connection starts with a plain text connection to the proxy server. After
         // the connection has been established with the proxy server, the connection might be
         // upgraded to TLS before we send our first request.
-        let bootstrap = self.makePlainBootstrap(deadline: deadline, eventLoop: eventLoop)
+        let bootstrap = self.makePlainBootstrap(requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop)
         return bootstrap.connect(host: proxy.host, port: proxy.port).flatMap { channel in
             let encoder = HTTPRequestEncoder()
             let decoder = ByteToMessageHandler(HTTPResponseDecoder(leftOverBytesStrategy: .dropBytes))
@@ -243,8 +257,9 @@ extension HTTPConnectionPool.ConnectionFactory {
         }
     }
 
-    private func makeSOCKSProxyChannel(
+    private func makeSOCKSProxyChannel<Requester: HTTPConnectionRequester>(
         _ proxy: HTTPClient.Configuration.Proxy,
+        requester: Requester,
         connectionID: HTTPConnectionPool.Connection.ID,
         deadline: NIODeadline,
         eventLoop: EventLoop,
@@ -253,7 +268,7 @@ extension HTTPConnectionPool.ConnectionFactory {
         // A proxy connection starts with a plain text connection to the proxy server. After
         // the connection has been established with the proxy server, the connection might be
         // upgraded to TLS before we send our first request.
-        let bootstrap = self.makePlainBootstrap(deadline: deadline, eventLoop: eventLoop)
+        let bootstrap = self.makePlainBootstrap(requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop)
         return bootstrap.connect(host: proxy.host, port: proxy.port).flatMap { channel in
             let socksConnectHandler = SOCKSClientHandler(targetAddress: SOCKSAddress(self.key.connectionTarget))
             let socksEventHandler = SOCKSEventsHandler(deadline: deadline)
@@ -331,14 +346,21 @@ extension HTTPConnectionPool.ConnectionFactory {
         }
     }
 
-    private func makePlainBootstrap(deadline: NIODeadline, eventLoop: EventLoop) -> NIOClientTCPBootstrapProtocol {
+    private func makePlainBootstrap<Requester: HTTPConnectionRequester>(
+        requester: Requester,
+        connectionID: HTTPConnectionPool.Connection.ID,
+        deadline: NIODeadline,
+        eventLoop: EventLoop
+    ) -> NIOClientTCPBootstrapProtocol {
         #if canImport(Network)
         if #available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *), let tsBootstrap = NIOTSConnectionBootstrap(validatingGroup: eventLoop) {
             return tsBootstrap
+                .channelOption(NIOTSChannelOptions.waitForActivity, value: self.clientConfiguration.networkFrameworkWaitForConnectivity)
                 .connectTimeout(deadline - NIODeadline.now())
                 .channelInitializer { channel in
                     do {
                         try channel.pipeline.syncOperations.addHandler(HTTPClient.NWErrorHandler())
+                        try channel.pipeline.syncOperations.addHandler(NWWaitingHandler(requester: requester, connectionID: connectionID))
                         return channel.eventLoop.makeSucceededVoidFuture()
                     } catch {
                         return channel.eventLoop.makeFailedFuture(error)
@@ -355,9 +377,17 @@ extension HTTPConnectionPool.ConnectionFactory {
         preconditionFailure("No matching bootstrap found")
     }
 
-    private func makeTLSChannel(deadline: NIODeadline, eventLoop: EventLoop, logger: Logger) -> EventLoopFuture<(Channel, String?)> {
+    private func makeTLSChannel<Requester: HTTPConnectionRequester>(
+        requester: Requester,
+        connectionID: HTTPConnectionPool.Connection.ID,
+        deadline: NIODeadline,
+        eventLoop: EventLoop,
+        logger: Logger
+    ) -> EventLoopFuture<(Channel, String?)> {
         precondition(self.key.scheme.usesTLS, "Unexpected scheme")
         let bootstrapFuture = self.makeTLSBootstrap(
+            requester: requester,
+            connectionID: connectionID,
             deadline: deadline,
             eventLoop: eventLoop,
             logger: logger
@@ -387,8 +417,13 @@ extension HTTPConnectionPool.ConnectionFactory {
         return channelFuture
     }
 
-    private func makeTLSBootstrap(deadline: NIODeadline, eventLoop: EventLoop, logger: Logger)
-        -> EventLoopFuture<NIOClientTCPBootstrapProtocol> {
+    private func makeTLSBootstrap<Requester: HTTPConnectionRequester>(
+        requester: Requester,
+        connectionID: HTTPConnectionPool.Connection.ID,
+        deadline: NIODeadline,
+        eventLoop: EventLoop,
+        logger: Logger
+    ) -> EventLoopFuture<NIOClientTCPBootstrapProtocol> {
         var tlsConfig = self.tlsConfiguration
         switch self.clientConfiguration.httpVersion.configuration {
         case .automatic:
@@ -408,11 +443,13 @@ extension HTTPConnectionPool.ConnectionFactory {
                 options -> NIOClientTCPBootstrapProtocol in
 
                 tsBootstrap
+                    .channelOption(NIOTSChannelOptions.waitForActivity, value: self.clientConfiguration.networkFrameworkWaitForConnectivity)
                     .connectTimeout(deadline - NIODeadline.now())
                     .tlsOptions(options)
                     .channelInitializer { channel in
                         do {
                             try channel.pipeline.syncOperations.addHandler(HTTPClient.NWErrorHandler())
+                            try channel.pipeline.syncOperations.addHandler(NWWaitingHandler(requester: requester, connectionID: connectionID))
                             // we don't need to set a TLS deadline for NIOTS connections, since the
                             // TLS handshake is part of the TS connection bootstrap. If the TLS
                             // handshake times out the complete connection creation will be failed.

Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool.swift

@@ -467,6 +467,16 @@ extension HTTPConnectionPool: HTTPConnectionRequester {
             $0.failedToCreateNewConnection(error, connectionID: connectionID)
         }
     }
+
+    func waitingForConnectivity(_ connectionID: HTTPConnectionPool.Connection.ID, error: Error) {
+        self.logger.debug("waiting for connectivity", metadata: [
+            "ahc-error": "\(error)",
+            "ahc-connection-id": "\(connectionID)",
+        ])
+        self.modifyStateAndRunActions {
+            $0.waitingForConnectivity(error, connectionID: connectionID)
+        }
+    }
 }
 
 extension HTTPConnectionPool: HTTP1ConnectionDelegate {

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift

@@ -241,6 +241,12 @@ extension HTTPConnectionPool {
             }
         }
 
+        mutating func waitingForConnectivity(_ error: Error, connectionID: Connection.ID) -> Action {
+            self.lastConnectFailure = error
+
+            return .init(request: .none, connection: .none)
+        }
+
         mutating func connectionCreationBackoffDone(_ connectionID: Connection.ID) -> Action {
             switch self.lifecycleState {
             case .running:

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift

@@ -406,6 +406,11 @@ extension HTTPConnectionPool {
             return .init(request: .none, connection: .scheduleBackoffTimer(connectionID, backoff: backoff, on: eventLoop))
         }
 
+        mutating func waitingForConnectivity(_ error: Error, connectionID: Connection.ID) -> Action {
+            self.lastConnectFailure = error
+            return .init(request: .none, connection: .none)
+        }
+
         mutating func connectionCreationBackoffDone(_ connectionID: Connection.ID) -> Action {
             // The naming of `failConnection` is a little confusing here. All it does is moving the
             // connection state from `.backingOff` to `.closed` here. It also returns the

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+StateMachine.swift

@@ -211,6 +211,14 @@ extension HTTPConnectionPool {
             })
         }
 
+        mutating func waitingForConnectivity(_ error: Error, connectionID: Connection.ID) -> Action {
+            self.state.modify(http1: { http1 in
+                http1.waitingForConnectivity(error, connectionID: connectionID)
+            }, http2: { http2 in
+                http2.waitingForConnectivity(error, connectionID: connectionID)
+            })
+        }
+
         mutating func connectionCreationBackoffDone(_ connectionID: Connection.ID) -> Action {
             self.state.modify(http1: { http1 in
                 http1.connectionCreationBackoffDone(connectionID)

Sources/AsyncHTTPClient/HTTPClient.swift

@@ -655,6 +655,10 @@ public class HTTPClient {
         /// is set to `.automatic` by default which will use HTTP/2 if run over https and the server supports it, otherwise HTTP/1
         public var httpVersion: HTTPVersion
 
+        /// Whether `HTTPClient` will let Network.framework sit in the `.waiting` state awaiting new network changes, or fail immediately. Defaults to `true`,
+        /// which is the recommended setting. Only set this to `false` when attempting to trigger a particular error path.
+        public var networkFrameworkWaitForConnectivity: Bool
+
         public init(
             tlsConfiguration: TLSConfiguration? = nil,
             redirectConfiguration: RedirectConfiguration? = nil,
@@ -671,6 +675,7 @@ public class HTTPClient {
             self.proxy = proxy
             self.decompression = decompression
             self.httpVersion = .automatic
+            self.networkFrameworkWaitForConnectivity = true
         }
 
         public init(tlsConfiguration: TLSConfiguration? = nil,

Sources/AsyncHTTPClient/NIOTransportServices/NWWaitingHandler.swift

@@ -0,0 +1,46 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the AsyncHTTPClient open source project
+//
+// Copyright (c) 2022 Apple Inc. and the AsyncHTTPClient project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of AsyncHTTPClient project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+#if canImport(Network)
+import Network
+import NIOCore
+import NIOHTTP1
+import NIOTransportServices
+
+@available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
+final class NWWaitingHandler<Requester: HTTPConnectionRequester>: ChannelInboundHandler {
+    typealias InboundIn = Any
+    typealias InboundOut = Any
+
+    private var requester: Requester?
+    private let connectionID: HTTPConnectionPool.Connection.ID
+
+    init(requester: Requester, connectionID: HTTPConnectionPool.Connection.ID) {
+        self.requester = requester
+        self.connectionID = connectionID
+    }
+
+    func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
+        if let waitingEvent = event as? NIOTSNetworkEvents.WaitingForConnectivity, let requester = self.requester {
+            requester.waitingForConnectivity(self.connectionID, error: HTTPClient.NWErrorHandler.translateError(waitingEvent.transientError))
+            self.requester = nil
+        }
+        context.fireUserInboundEventTriggered(event)
+    }
+
+    func handlerRemoved(context: ChannelHandlerContext) {
+        self.requester = nil
+    }
+}
+#endif

Tests/AsyncHTTPClientTests/HTTP2ConnectionTests.swift

@@ -405,6 +405,10 @@ extension TestConnectionCreator: HTTPConnectionRequester {
         }
         wrapper.fail(error)
     }
+
+    func waitingForConnectivity(_: HTTPConnectionPool.Connection.ID, error: Swift.Error) {
+        preconditionFailure("TODO")
+    }
 }
 
 class TestHTTP2ConnectionDelegate: HTTP2ConnectionDelegate {

Tests/AsyncHTTPClientTests/HTTPClient+SOCKSTests.swift

@@ -90,8 +90,10 @@ class HTTPClientSOCKSTests: XCTestCase {
     }
 
     func testProxySOCKSBogusAddress() throws {
+        var config = HTTPClient.Configuration(proxy: .socksServer(host: "127.0.."))
+        config.networkFrameworkWaitForConnectivity = false
         let localClient = HTTPClient(eventLoopGroupProvider: .shared(self.clientGroup),
-                                     configuration: .init(proxy: .socksServer(host: "127.0..")))
+                                     configuration: config)
 
         defer {
             XCTAssertNoThrow(try localClient.syncShutdown())
@@ -102,8 +104,11 @@ class HTTPClientSOCKSTests: XCTestCase {
     // there is no socks server, so we should fail
     func testProxySOCKSFailureNoServer() throws {
         let localHTTPBin = HTTPBin()
+        var config = HTTPClient.Configuration(proxy: .socksServer(host: "localhost", port: localHTTPBin.port))
+        config.networkFrameworkWaitForConnectivity = false
+
         let localClient = HTTPClient(eventLoopGroupProvider: .shared(self.clientGroup),
-                                     configuration: .init(proxy: .socksServer(host: "localhost", port: localHTTPBin.port)))
+                                     configuration: config)
         defer {
             XCTAssertNoThrow(try localClient.syncShutdown())
             XCTAssertNoThrow(try localHTTPBin.shutdown())
@@ -113,8 +118,11 @@ class HTTPClientSOCKSTests: XCTestCase {
 
     // speak to a server that doesn't speak SOCKS
     func testProxySOCKSFailureInvalidServer() throws {
+        var config = HTTPClient.Configuration(proxy: .socksServer(host: "localhost"))
+        config.networkFrameworkWaitForConnectivity = false
+
         let localClient = HTTPClient(eventLoopGroupProvider: .shared(self.clientGroup),
-                                     configuration: .init(proxy: .socksServer(host: "localhost")))
+                                     configuration: config)
         defer {
             XCTAssertNoThrow(try localClient.syncShutdown())
         }
@@ -124,8 +132,11 @@ class HTTPClientSOCKSTests: XCTestCase {
     // test a handshake failure with a misbehaving server
     func testProxySOCKSMisbehavingServer() throws {
         let socksBin = try MockSOCKSServer(expectedURL: "/socks/test", expectedResponse: "it works!", misbehave: true)
+        var config = HTTPClient.Configuration(proxy: .socksServer(host: "localhost", port: socksBin.port))
+        config.networkFrameworkWaitForConnectivity = false
+
         let localClient = HTTPClient(eventLoopGroupProvider: .shared(self.clientGroup),
-                                     configuration: .init(proxy: .socksServer(host: "localhost", port: socksBin.port)))
+                                     configuration: config)
 
         defer {
             XCTAssertNoThrow(try localClient.syncShutdown())